Hadoop、Spark、Flink Shuffle对比

news2025/3/18 16:05:25

一、Hadoop的shuffle

前置知识：

Map任务的数量由Hadoop框架自动计算，等于分片数量，等于输入文件总大小 / 分片大小，分片大小为HDFS默认值128M，可调

Reduce任务数由用户在作业提交时通过Job.setNumReduceTasks(int)设置

数据分配到Reduce任务的时间点，在Map任务执行期间，通过Partitioner（分区器）确定每个键值对的目标Reduce分区。默认采取partition=hash(key) % numReduceTasks策略

Shuffle过程：

hadoop会先将map数据写入缓冲区，缓冲区达到某个阈值后，会把数据溢写至磁盘，溢写磁盘时会根据先将数据写入相应分区文件，进行排序

溢写完毕后，会将多个分区文件合并，再进行归并排序

Reduce任务主动从所有Map任务的磁盘中拉取（Pull）属于自己分区的数据，拉取到数据后，还会进行一次归并排序

可以看到一共进行了三次排序。这一设计是后来所有分布式计算框架混洗任务的基石。

QA：为什么Hadoop需要三次排序？

第一次排序是为了第二次归并排序方便

第二次归并排序是为了给reduce任务时，reduce任务可以顺序读

第三次排序是因为hadoop要保证同一个reduce的输出是有序的，同时如果输入的key是有序的，reduce处理完输出即可，如果是无序的，那么reduce需要保存再重排序，对于数据量大的场景容易oom

二、Spark的shuffle

前置知识：

map个数由Saprk分区数计算或者自定义，reduce个数由用户指定，如果没指定，通常是机器核数

map和reduce数据的交互方式依旧是，map后把数据写入文件中，reduce从文件中读取数据

分区ID是数据在Shuffle过程中被分配到的目标Reduce任务的编号，决定了数据最终由哪个Reduce任务处理。

计算方式：
默认使用HashPartitioner，根据Key的哈希值对Reduce任务数取模：

分区ID=hash(key) % numReduceTasks分区ID=hash(key) % numReduceTasks

2.1 哈希混洗

Spark 1.2 之前默认的Shuffle机制

map输出的数据不再排序，若有M个map任务和R个reduce任务，每个map任务生成R个文件，每个reduce任务拉取属于自己的文件

这样导致文件句柄数太多了，若M=1000、R=1000，则生成 1,000,000个文件，同时内存压力也比较大，如果需要排序要在reduce端把一个key的所有数据全部加载，所以后面使用了sort混洗

2.2 sort 混洗

Spark 1.2 引入，逐步成为默认机制

1. Map任务处理输入数据，生成<Key, Value>对，并按分区ID暂存到内存缓冲区

2. 当缓冲区达到阈值（如spark.shuffle.spill.numElementsForceSpillThreshold默认值）时，开始排序。

排序规则：
- 仅按分区ID排序（默认）：将数据按分区ID排序，同一分区内的数据无序。
- 按分区ID + Key排序（需配置）：
  若设置spark.shuffle.sort.byKey=true，则按(分区ID, Key)排序，同一分区内的数据按键有序。

3. 排序后的数据按分区ID顺序写入磁盘，生成一个临时溢写文件

4. Map任务结束时，将所有临时溢写文件合并为单个数据文件（data）和一个索引文件（index）

合并方式：
- 多路归并排序：将多个已按分区ID（或Key）排序的溢写文件合并，保持全局有序性。
- 索引文件生成：记录每个分区ID在数据文件中的起始和结束偏移量。

5. Reduce任务向Driver查询所有Map任务生成的数据文件和索引文件的位置

6. 若Map端已按Key排序，Reduce任务直接对多个有序数据块进行归并，生成全局有序数据集。

内存与磁盘结合：
- 数据量较小时，直接在内存中归并。
- 数据量较大时，使用外排序（溢出到磁盘，分批次归并

感觉这样下来，跟hadoop的shuffle就有点像了，这样有个好处是，map生成的文件就只有两个了，最终的文件就是 2 * R个

2.3 Spark和Hadoop shuffle的内存使用上的不同之处

Hadoop写文件时，是设置了一个内存阈值，到达了该阈值就会把内存内容写入文件中，比如阈值是80M，一个200M文件就要溢写三次，且缓冲区大小不可动态调整，无法根据任务需求扩展或收缩。

Spark 将内存划分为 存储内存（Storage Memory） 和 执行内存（Execution Memory），两者可动态借用，

Map 任务将数据按分区ID（或 Key）缓存在内存中。
溢出到磁盘：若内存不足，部分数据排序后写入磁盘临时文件。
合并最终文件：Map 结束时合并内存和磁盘数据，生成一个数据文件和一个索引文件。

举个spark处理数据的例子，假设有200MB数据：

(1) 内存排序

Map 任务处理数据后，先将键值对缓存在内存中，并按 分区ID（和可选的 Key）排序。
假设可用执行内存为 150MB，前 150MB 数据在内存中完成排序，生成一个 有序的内存块。

(2) 溢出到磁盘

当内存不足时，Spark 将内存中已排序的 150MB 数据 溢写到磁盘，生成一个临时文件（如 spill1），该文件内部保持有序。
剩余 50MB 数据继续在内存中排序，直到 Map 任务结束。

在 Map 任务结束时，所有内存和磁盘上的数据会被合并为一个全局有序的输出文件。具体流程如下：

假设 Map 任务生成以下两个有序片段：

内存块（150MB）：[A, B, D, F]
溢写文件（50MB）：[C, E, G]

归并过程：

初始化指针：内存块指向 A，溢写文件指向 C。
比较当前元素，选择最小者：
- 第一轮：A（内存块） → 写入最终文件。
- 第二轮：B（内存块） → 写入最终文件。
- 第三轮：C（溢写文件） → 写入最终文件。
- ...
最终合并结果：[A, B, C, D, E, F, G]。