由于内存不足，Spark Join 失败

Posted 2023-04-18

【中文标题】由于内存不足，Spark Join 失败

【英文标题】：Spark Join failed due to Out-Of-Memory

【发布时间】：2015-07-23 19:01:21

【问题描述】：

我的集群： 9 个从站，每个具有 100GB 内存和 320GB 硬盘驱动器。每台主机有 16 个内核。我在每个主机上启动了 15 个 spark 执行器，因此每个执行器的可用内存为 6GB。 我的应用：

val rdd1 = sc.textFile("a big file in S3. about 200GB" with 14M rows)
val rdd2 = sc.textFile("another big file in S3. about 200GB" with 14M rows)
val par = new HashPartitioner(150)
val rdd1Paired = rdd1.map(regular expression to get one string from each row).filter(non matched rows).partitionBy(par)
val rdd2Paired = rdd2.mpa(regular expression to get one string from each row).filter(non matched rows).partitionBy(par)
val rdd3 = rdd1.join(rdd2, par)
rdd3.count()

我从 spark UI 中了解到，该作业被安排在三个阶段。过滤 rdd1，过滤 rdd2 和计数。虽然filtering1和filtering2成功，但由于OOM，count总是失败。奇怪的是，这项工作总是在计数阶段（149/150）挂起。我检查了分配了 TID150 的执行程序，发现 Shuffle 读取显着增加。由于OOM，它会在一段时间后崩溃。我还看到 GC 在该执行程序上经常发生。

这里的问题： 为什么只有一个执行器正在获取所有数据（我检查了执行器所做的最后一个操作是开始获取工作）？据我了解，一旦我对两个 RDD 使用相同的分区器，它们将被共同分区。并在同一个工作中启动它们可以保证数据的协同定位。所以加入应该发生在每个执行者本身。前 149 个任务很快退出，似乎什么也没做。看起来最后一个任务是尝试所有的工作。

【参考方案1】：

键的分布不均匀。检查以确保您的任何过滤器功能都不会附带导致某些键更集中的影响。

【讨论】：

我检查了代码，发现问题是由大量重复键引起的。

【参考方案2】：

我猜您从数据中生成的密钥分布是不正确的，因此其中很多最终都位于同一个分区中。

它是最后一个失败的任务的事实是它是最大的，所以它执行的时间最长并且是最后一个。

为了解决这个问题，增加分区的数量和/或加强你的服务器

【讨论】：

感谢您的回复。我确实在我的哈希键上测试了 HashPartitioner(5000)（从同一个文件中提取的完全相同的键列表）。但是，分布看起来不错。我正在使用分区 10K 运行另一个作业，看看它是否失败。