Spark OOM 错误解释及缓解

Question

有时，你会得到一个 OutOfMemoryError 不是因为你的 RDD 不适合内存，而是因为你的一个任务的工作集，比如 groupByKey 中的一个 reduce 任务，太大了。 Spark 的 shuffle 操作（sortByKey、groupByKey、reduceByKey、join 等）在每个任务中构建一个哈希表来执行分组，这通常可能很大。这里最简单的解决方法是提高并行度，使每个任务的输入集更小。

我是这么认为的，如果我错了，请纠正我。

假设有 2 个数据节点来处理数据集，并且这两个节点共同拥有 32GB 的内存（每个数据节点 16GB）。数据集大小为 100 GB，让我们假设这些数据在被 spark 读取时，被划分为 10 个分区，每个分区 10GB。很明显，100GB 的文件一次无法放入 32GB 的 RAM。因此必须将分区加载到内存中并以迭代方式进行处理。所以我假设如下。

第一次迭代，2 个分区，每个 10GB 被加载到每个数据节点的内存中。 第二次迭代，2 个分区，每个 10GB 被加载到每个数据节点的内存中。 …… …… 第五次迭代，2 个分区，每个 10GB 被加载到每个数据节点上的内存中。

如果这是 spark 的处理方式，则在每次迭代期间，只有 2 个分区被加载到内存中。这是否意味着其他无法容纳在内存中的分区被读取但溢出到磁盘并且它们正在等待内存被释放？或者根本不读取这些分区，只有在资源可用时才会读取它们。哪个是真的？

在处理过程中，如果需要 groupby/reduceby/join，那么它会强制进行 shuffle。因此，如果其中一个 shuffle 分区大于 RAM 大小，则作业将因 OOM 错误而失败。例如，10 个分区被处理和洗牌。现在 shuffle 分区只有 4 个分区，每个分区 25GB。 （默认 shuffle 分区是 200 个，但只有 4 个分区的总数据是空的。）由于 shuffle 分区大小大于 16MB RAM，spark 作业会失败吗？我的理解正确吗？

我明白，您并不需要您的数据适合内存。 Spark 在分区的基础上处理数据。但我的问题是，如果分区本身不适合内存怎么办。它还会将数据溢出到磁盘并开始处理，还是会因 OOM 错误而失败？

我的第二个问题是，如果在执行上述 spark 作业（job1）期间触发了另一个 spark 作业（Job2），并假设这还有 100GB 文件要处理，每个分区 10GB。因此，当 job1 Iteration1 正在执行时，内存中只有 6 MB 可用插槽。 job2 的 10GB 分区无法加载到内存中来处理 job2。那么 Job2 会等到内存被释放吗？还是该作业也会因 OOM 错误而失败？

Answer 1

解释（粗体）是正确的。

在你的 cmets 上：

• 除非您明确重新分区，否则您的分区将与 HDFS 块大小相关，即 128MB 大小以及构成该文件的数量。

• 然后你有多个执行者，比如 2 个，每个 Worker / Data Node。然后最多 4 个任务/分区将在任何给定时间处于活动状态。

  • 如果最多可以服务 4 个分区，那么将所有分区加载到内存有什么意义？您会阻塞系统，从而损害其他 Spark 应用程序。这一切就像一个普通的操作系统。

当然它有点复杂，例如如果你有 10 个数据节点并且只分配了 2 个执行器，那么就会有流量可以移动。保持简单。

仅当分区超过最大分区大小时才会发生 OOM 错误。剩余的磁盘空间需要用于溢出。