Spark 任务内存分配

Posted 2023-04-18

【中文标题】Spark 任务内存分配

【英文标题】：Spark Task Memory allocation

【发布时间】：2017-08-07 18:28:10

【问题描述】：

我正在尝试找出在集群节点上配置内存的最佳方式。但是，我相信为此我需要进一步了解一些事情，例如 spark 如何处理跨任务的内存。

例如，假设我有 3 个执行器，每个执行器最多可以并行运行 8 个任务（即 8 个核心）。如果我有一个有 24 个分区的 RDD，这意味着理论上所有的分区都可以并行处理。但是，如果我们在这里放大一个执行器，这假设每个任务都可以在内存中拥有它们的分区以对它们进行操作。如果不是，则意味着不会发生 8 个并行任务，需要进行一些调度。

因此我得出的结论是，如果有人寻求真正的并行性，对分区的大小有所了解会有所帮助，因为它会告诉您如何调整执行程序的大小以实现真正的并行性。

Q0 - 我只是想更好地理解一点，如果不这样做会发生什么 所有分区都可以容纳在一个执行程序中的内存中吗？有一些 溢出在磁盘上，而其他人在内存中进行操作？火花 为每个任务预留内存，如果检测到没有 够了，它会安排任务吗？或者做一个简单地跑出去 内存错误。

Q1 - 真正的并行性 执行器还取决于执行器上可用的内存量？ 换句话说，我的集群中可能有 8 个核心，但如果我没有 足够的内存一次加载我的数据的 8 个分区，那么我不会 完全并行。

作为最后一点，我多次看到以下声明，但发现它有点混乱：

“增加分区数量也有助于减少 内存不足错误，因为这意味着 Spark 将在 每个执行者的数据的较小子集。”

这究竟是如何工作的？我的意思是 spark 可能适用于较小的子集，但如果总的分区集无论如何都不能放入内存，会发生什么？

【参考方案1】：

为什么要增加任务（分区）的数量？

我想先回答最后一个让您感到困惑的问题。 这是另一个question的引述：

Spark 不需要将所有内容加载到内存中即可对其进行处理。这是因为 Spark 会将数据划分为更小的块并分别对这些块进行操作。

事实上，默认情况下，Spark 会尝试to split input data automatically 进入一些最佳分区数：

Spark 会根据文件的大小自动设置要在每个文件上运行的“映射”任务的数量

可以指定正在执行的操作的分区数（例如cogroup：def cogroup[W](other: RDD[(K, W)], numPartitions: Int)），还可以在任何 RDD 转换后执行.repartition()。

此外，在文档的同一段落中，他们说：

一般来说，我们建议集群中每个 CPU 内核执行 2-3 个任务。

总结：

默认分区数是一个好的开始； 通常建议每个 CPU 2-3 个分区。

Spark 如何处理不适合内存的输入？

简而言之，通过对输入和中间结果 (RDD) 进行分区。通常每个小块都适合执行程序可用的内存并且可以快速处理。

Spark 能够计算出caching the RDDs。默认情况下，每次重用 RDD 时都会重新计算（不缓存）；调用 .cache() 或 .persist() 可以帮助将已计算的结果保存在内存中或磁盘上。

每个执行器内部都有一个在执行和存储之间浮动的内存池（有关详细信息，请参阅here）。当没有足够的内存来执行任务时，Spark 首先会尝试驱逐一些存储缓存，然后将任务数据溢出到磁盘上。有关详细信息，请参阅这些slides。这个blog post很好地描述了执行器和存储内存之间的平衡，它也有一个很好的说明：

OutOfMemory 通常不是直接因为输入数据大而直接发生，而是因为分区不佳，因此辅助数据结构很大，例如 reducer 上的HashMap（here 文档再次建议分区多于执行器）。所以，不，OutOfMemory 不会仅仅因为大输入而发生，但处理起来可能非常慢（因为它必须从磁盘写入/读取）。他们还建议使用小至 200 毫秒（运行时间）的任务对于 Spark 来说是可以的。

大纲：正确拆分数据：每个核心超过1个分区，每个任务的运行时间应>200毫秒。默认分区是一个很好的起点，手动调整参数。

（我建议在 1/8 集群上使用输入数据的 1/8 子集来找到最佳分区数。）

同一个executor内的任务会互相影响吗？

简短回答：确实如此。有关更多详细信息，请查看我上面提到的slides（从幻灯片 #32 开始）。

所有 N 个任务都获得第 N 部分可用内存，因此会影响彼此的“并行性”。如果我很好地解释了您对真正的并行性的想法，那就是“充分利用 CPU 资源”。在这种情况下，是的，小的内存池会导致磁盘上的数据溢出，并且计算成为 IO 密集型（而不是 CPU 密集型）。

进一步阅读

我强烈推荐整个章节Tuning Spark 和Spark Programming Guide。另请参阅 Alexey Grishchenko 在Spark Memory Management 上的这篇博文。

【讨论】：

非常感谢您的详细回答。我会经历这一切，然后会回复你。一条评论，我红色 Deep Dive 内存管理，似乎暗示的是，当没有足够的 RAM 时，spark 将与 DISK 一起工作，所以从某种意义上说，这意味着 Executor 可用的内存也应该考虑为那个 Executor 的磁盘空间！我仍然不确定是否可以简单地决定在节点上加载任意数量的数据并对其进行操作。

是的，确切地说，多余的数据会溢出到磁盘上。一般来说，我希望避免 Spark 溢出到磁盘上，因为它会使执行速度变慢。

伟大的洞察力谢谢！可能导致 OOM 的“辅助”JVM 数据结构是否会随着分区的记录数或其大小（以字节为单位）而增长？也许您对这两种类型都有示例？