当使用Spark进行计算时，我们经常遇到Job Out Of Memory（OOM），其中很大一部分发生在Shuffle阶段。所以在Spark Shuffle中，使用更多内存和OOM可能在哪里？为此，本文将重点讨论与内存使用相关的Spark内存管理和Shuffle过程中的上述问题；然后，简要分析Spark Shuffle中可能有OOM的原因。
一，Spark内存管理和消费模型
在分析Spark Shuffle内存使用情况之前。我们首先理解以下问题：当一个Spark子任务分配给Executor时，Spark管理内存和消耗内存的一般模型是什么？ （注意：由于OOM主要发生在Executor端，因此它是连接的。下面的讨论主要用于内存管理和Executor端的使用。
1，在Spark中，使用抽象类MemoryConsumer来表示需要使用内存的消费者。在此类中定义了一些用于分配，释放和溢出内存数据到磁盘的方法或接口。特定的消费者可以继承MemoryConsumer来实现特定的行为。因此，在执行Spark任务期间，将存在具有不同类型和不同数量的各种类型的特定消费者。 Spark Shuffle中使用的ExternalAppendOnlyMap，ExternalSorter等（稍后会详细介绍）。
2，MemoryConsumer将适用于释放相关内存到TaskMemoryManager来执行。分配Spark任务以在Executor上运行时，将创建TaskMemoryManager。在TaskMemoryManager执行内存分配之前，您需要首先应用于MemoryManager，然后TaskMemoryManager使用MemoryAllocator执行实际的内存分配。3，Executor中的MemoryManager将管理内存的使用。由于每个TaskMemoryManager在执行实际内存分配之前请求MemoryManager。因此，MemoryManager对使用内存的当前进程有全面的了解。
MemoryManager，TaskMemoryManager和MemoryConsumer之间的对应关系，如下所示。通常，MemoryManager对应于至少一个TaskMemoryManager（由executor-core参数指定），而TaskMemoryManager对应于多个MemoryConsumers（取决于任务）。
一篇关于Spark Shuffle内存使用情况的文章
在了解了上面的内存消耗的整个过程之后，有两个问题需要注意：
1.当Executor上同时执行多个Tasks时，将有多个TaskMemoryManagers共享MemoryManager管理的内存。那么MemoryManager是如何分配的呢？答案是每个任务可以分配的内存范围是[1 /（2 * n），1/n]，其中n是正在运行的任务数。因此，多个并发运行的任务将使每个任务的内存可用更小。
2，如前所述，MemoryConsumer中有一个Spill方法。当MemoryConsumer没有足够的内存时，Spill可以将当前内存存储到磁盘，从而避免不受控制的内存使用。但是，堆内存的应用和释放实际上是由JVM管理的。因此，在计算堆中内存的具体用法时，考虑到性能和其他原因，Spark当前使用抽样统计方法来计算MemoryConsumer使用的内存，因此堆中内存的实际使用量并不特别准确。因此，OOM可能是由于无法及时溢出造成的。
二，Spark Shuffle流程
Spark Shuffle的整个过程如下，分为两个阶段：Shuffle Write和Shuffle Read。Write阶段通常进行排序（最低要求是按分区排序），可能的聚合和合并（当磁盘上有多个文件时），最后每个write Task将生成数据并索引两个文件。其中，数据文件根据分区存储，即同一分区的数据在文件中是连续的，索引文件记录文件中每个分区的起始位置和结束位置。
对于Shuffle Read，您可能首先需要通过网络从每个Write任务节点获取给定分区的数据，即数据文件的连续段，然后排序，合并等，最后形成一个计算结果。
一篇关于Spark Shuffle内存使用情况的文章
对于Shuffle Write，Spark目前有三种实现，即BypassMergeSortShuffleWriter，UnsafeShuffleWriter和SortShuffleWriter（具体实现具有判断条件，此处未显示）。 Shuffle Read只有一个实现。
2.1随机写阶段分析
2.1.1 BypassMergeSortShuffleWriter分析
对于BypassMergeSortShuffleWriter的实现，一般的实现过程是首先为每个分区创建一个临时分区文件，将数据写入相应的分区文件，最后将所有分区文件合并为一个数据文件，索引文件为产生。由于此过程不进行排序，因此组合（如果需要组合不使用此实现）等等，因此对于BypassMergeSortShuffleWriter，总体而言，它不是非常占用内存。
2.1.2 SortShuffleWriter分析
SortShuffleWriter是最常见的实现，是日常使用最频繁的实现。 SortShuffleWriter主要委托ExternalSorter进行数据插入，排序，合并，组合，最后写入数据和索引文件。 ExternalSorter实现前面提到的MemoryConsumer接口。让我们分析每个进程中内存的使用：
1，对于数据写入，根据是否需要进行组合，数据将被插入到PartitionedAppendOnlyMap Map或PartitionedPairBuffer数组中。每隔一段时间，当MemoryManager没有足够的可用内存，或者当数据量超过spark.shuffle.spill.numElementsForceSpillThreshold的阈值时（默认值为Long的最大值，它不起作用），Spill内存数据将被发送到文件。假设您可以连续应用于内存，则Write阶段中的所有数据都将保留在内存中。可以看出PartitionedAppendOnlyMap或PartitionedPairBuffer是更多的内存。2，无论是PartitionedAppendOnlyMap还是PartitionedPairBuffer，使用的排序算法都是TimSort。在正常使用下使用的临时额外空间很小，但在最坏的情况下是n/2，其中n是要排序的数组的长度（有关详细信息，请参阅TimSort实现）。
3.由于应用程序没有足够的内存而插入数据时，Spill数据将被发送到磁盘。在将最终排序结果写入数据文件之前，需要合并内存中的PartitionedAppendOnlyMap或PartitionedPairBuffer以及已溢出到磁盘的SpillFiles。合并的一般过程如下所示。
一篇关于Spark Shuffle内存使用情况的文章
从上图可以看出，它几乎是排序和排序的过程，这表明这个过程没有太多额外的内存消耗。合并过程中的聚合计算通常是类似的过程。唯一需要注意的是键值冲突的情况，即当前输入的队列的键值的哈希值是相同的，但实际键值不相等。在这种情况下，需要额外的空间来保持所有键值不同，但哈希值与中间结果相同。但总的来说，发生这种情况的可能性并不是特别大。
4，写入数据文件的过程涉及不同数据流之间的转换，并且在写入流的过程中，通常有一个缓存，主要由参数spark.shuffle.file.buffer和spark.shuffle.spill控制。 batchSize一般来说，这部分成本并不大。
上面分析了SortShuffleWriter写阶段的主要过程，可以看出主内存消耗是写入PartitionedAppendOnlyMap或PartitionedPairBuffer阶段的。
2.1.3 UnsafeShuffleWriter
UnsafeShuffleWriter是SortShuffleWriter的优化，一般类似于SortShuffleWriter，这里不再赘述。从内存使用的角度来看，主要区别如下：
一方面，在SortShuffleWriter的PartitionedAppendOnlyMap或PartitionedPairBuffer中，存储的特定类型的键值或值，即Java对象，是反序列化后的数据。在UnsafeShuffleWriter的ShuffleExternalSorter中，数据被序列化并存储在实际页面中，并且在数据写入过程中另外写入长度信息。总的来说，序列化后数据的大小远小于序列化之前的数据。另一方面，UnsafeShuffleWriter需要一个额外的存储记录（LongArray），它保存分区信息和指向序列化数据的指针（编码的页面数和偏移量）。 UnsafeShuffleWriter中此部分存储的开销与SortShuffleWriter相关。
2.2随机读取阶段分析
Spark Shuffle Read主要经历诸如获取数据，序列化流，添加指标统计，可能的聚合（聚合）计算和排序等过程。一般过程如下所示。
一篇关于Spark Shuffle内存使用情况的文章

• 上一篇: 蓝盟IT外包，元宇宙：四梁八柱是什么
  下一篇: 使用stunnel来保护telnet连接

• 分享到：