为什么要增加任务数量(分区)?
首先,我想回答让您感到困惑的最后一个问题。以下是来自另一个问题的引用:
Spark不需要将所有数据加载到内存中才能处理它。这是因为Spark会将数据分区成较小的块并分别处理。
事实上,默认情况下,Spark尝试将输入数据自动拆分为一些最优分区:
Spark会根据文件大小自动设置要在每个文件上运行的“map”任务数。
可以指定正在执行的操作的分区数(例如对于: def cogroup [W](other:RDD [(K,W)],numPartitions:Int)
),并且在任何RDD转换之后进行.repartition()
。
此外,在文档的同一段落中,他们还说:
通常,我们建议在集群中的每个CPU核心上运行2-3个任务。
总而言之:
- 默认分区数是一个很好的起点;
- 通常建议每个CPU使用2-3个分区。
Spark如何处理内存不足的输入?
简而言之,通过对输入和中间结果(RDD)进行分区。通常,每个小块都适合执行器可用的内存,并且可以快速处理。
Spark能够缓存计算过的RDD。默认情况下,每当RDD被重复使用时,它将被重新计算(未缓存);调用.cache()
或.persist()
可帮助在内存或磁盘上保持已经计算的结果。
内部每个执行器都有一个内存池,在执行和存储之间浮动(有关详细信息,请参见此处)。当任务执行时没有足够的内存时,Spark首先尝试清除一些存储缓存,然后将任务数据溢出到磁盘上。有关更多详细信息,请参见这些幻灯片。在这篇博客文章中,介绍了执行器和存储内存之间的平衡,并提供了漂亮的插图:
![Spark memory allocation](https://istack.dev59.com/KXLWD.webp)
内存溢出通常并不是由于大数据量直接引起的,而是由于拆分不当导致的大型辅助数据结构(例如Reducer上的HashMap)。因此,OutOfMemory不会仅因为输入数据太大而发生,但它可能会处理得非常缓慢(因为必须从磁盘读取/写入)。文档建议在Reducer上使用比执行器更多的分区。此外,他们建议对于Spark,200ms(运行时间)左右的任务大小是可以接受的。
概述:适当地拆分您的数据:每个核心超过1个分区,每个任务的运行时间应> 200毫秒。默认分区是一个很好的起点,可以手动调整参数。
(我建议在1/8集群上使用1/8输入数据子集来找到最佳分区数。)
同一执行程序中的任务是否会相互影响?
简短回答:他们会。更多详情请查看我上面提到的幻灯片(从第32张幻灯片开始)。
所有N个任务都会获得可用内存的第N部分,因此会影响彼此的“并行性”。如果我正确理解了您对真正的并行性的想法,那就是“充分利用CPU资源”。在这种情况下,是的,小内存池将导致数据溢出到磁盘上,从而使计算过程变得I/O限制(而不是CPU限制)。
进一步阅读
我强烈建议阅读整章《调整Spark》和Spark编程指南。还请查看Alexey Grishchenko关于Spark内存管理的博客文章。