Shuffle是MapReduce最贵的操作——Shuffle过程详解和优化

# Map端Shuffle的详细过程
#
# 1. Map Task处理输入数据，产生Key-Value输出
#
# 2. 输出写入环形缓冲区（默认100MB）
#    mapreduce.task.io.sort.mb = 100
#
# 3. 在缓冲区内按Key排序 + 按Reduce分区号分组
#    分区器决定每个Key去哪个Reduce（默认Hash分区）
#
# 4. 缓冲区使用率达到80%时触发溢写（Spill）
#    mapreduce.map.sort.spill.percent = 0.8
#    溢写到本地磁盘（一个排好序的临时文件）
#
# 5. 如果配了Combiner，溢写前先做本地聚合
#    比如WordCount场景，相同的Word先本地求和
#    减少写入磁盘和网络传输的数据量
#
# 6. Map Task完成后，合并所有溢写文件
#    多路归并排序（Merge Sort），生成一个有序的大文件
#    这个文件按Reduce分区号分段存储

# Reduce端Shuffle的详细过程
#
# 1. Reduce Task启动后，从所有Map Task拉取属于自己分区的数据
#    通过HTTP从各Map节点下载
#    mapreduce.reduce.shuffle.parallelcopies = 5（并行拉取线程数）
#
# 2. 拉取到的数据先放内存缓冲区
#    mapreduce.reduce.shuffle.input.buffer.percent = 0.7
#    内存不够就溢写到磁盘
#
# 3. 所有数据拉取完毕后，做多路归并排序
#    把来自不同Map的有序数据合并成一个全局有序的流
#
# 4. 按Key分组，交给Reduce函数处理
#    相同Key的所有Value被聚合
#
# 5. Reduce输出写入HDFS（最终结果）

#  Map Task 1     Map Task 2     Map Task 3
#  ┌────────┐    ┌────────┐    ┌────────┐
#  │排序+溢写│    │排序+溢写│    │排序+溢写│
#  │+Combiner│    │+Combiner│    │+Combiner│
#  └───┬────┘    └───┬────┘    └───┬────┘
#      │Spill文件    │Spill文件    │Spill文件
#      │合并          │合并          │合并
#      ↓              ↓              ↓
#  ┌────────────────────────────────────┐
#  │         网络传输（HTTP拉取）         │
#  └────────────────────────────────────┘
#      ↓              ↓
#  Reduce Task 1  Reduce Task 2
#  ┌────────┐    ┌────────┐
#  │归并排序  │    │归并排序  │
#  │+聚合    │    │+聚合    │
#  │写HDFS   │    │写HDFS   │
#  └────────┘    └────────┘

# 关键调优参数
 
# Map端缓冲区（默认100MB，调大减少溢写次数）
mapreduce.task.io.sort.mb = 200
 
# Map输出压缩（减少网络传输）
mapreduce.map.output.compress = true
mapreduce.map.output.compress.codec = org.apache.hadoop.io.compress.SnappyCodec
 
# Reduce端并行拉取线程数（默认5，调大加速拉取）
mapreduce.reduce.shuffle.parallelcopies = 10
 
# Reduce数量（太少导致单个Reduce压力大，太多产生小文件）
mapreduce.job.reduces = 100
# 经验公式：Reduce数 = 总Shuffle数据量 / 256MB

Combiner——Shuffle优化的利器

Combiner是Map端的'本地Reduce'——在数据写入磁盘前先做一次局部聚合，减少Shuffle的数据量。

// WordCount示例中的Combiner效果
// 假设Map Task 1处理了1000行文本，输出10万个(word, 1)对
 
// 不用Combiner：10万个(word, 1)全部Shuffle到Reduce
// 用Combiner：先本地聚合，(hello, 500)(world, 300)...
//   只需要Shuffle几千个聚合后的结果
 
// Combiner的使用条件：
// 操作必须是可交换可结合的（Commutative + Associative）
// SUM, COUNT, MAX, MIN 可以用Combiner
// AVG, MEDIAN 不能直接用（需要特殊处理）
 
// 设置Combiner
job.setCombinerClass(MyReducer.class);  // 通常和Reducer相同

实战经验：Shuffle数据量减少45%

在某知名出海硬件品牌，有一个Spark ETL作业处理设备埋点数据。原始SQL是SELECT * FROM events JOIN devices ON ...，Shuffle数据量巨大（几十GB）。

优化方案：① 把SELECT *改成只选需要的15个字段（原表有50个字段），Shuffle数据量减少了45%；② JOIN之前加过滤条件（WHERE dt = 'xxx' AND event_type IN ('activate', 'purchase')），输入数据量减少60%；③ 开启Shuffle压缩（spark.shuffle.compress = true + Snappy编码）。

效果：作业从40分钟降到18分钟，Shuffle数据量从50GB降到12GB。其中列裁剪贡献最大（45%），过滤前置其次，压缩锦上添花。

教训：SELECT *是Shuffle优化的大敌——你Shuffle了一堆最终不需要的列。这个看起来简单的优化，在生产环境中效果最明显、ROI最高。

面试高频追问

追问一：MapReduce的Shuffle和Spark的Shuffle有什么区别？

原理类似（都是按Key重分布数据），但实现不同：MR的Shuffle必须全量排序+落盘，Spark有Sort-Based和Hash-Based两种Shuffle，且中间数据可以利用内存。Spark 2.0+默认用Sort-Based Shuffle，在大多数场景下比MR的Shuffle更高效。

追问二：什么操作会触发Shuffle？

在Spark中：groupByKey、reduceByKey、join、distinct、repartition、sort等操作都会触发Shuffle。在SQL中：GROUP BY、JOIN、DISTINCT、ORDER BY、窗口函数都涉及Shuffle。优化的核心是减少Shuffle的次数和每次Shuffle的数据量。

追问三：数据倾斜本质上是Shuffle的什么问题？

数据倾斜是Shuffle分区不均匀——某些Key的数据量远大于其他Key，导致处理这些Key的Task运行时间是其他的几倍甚至几十倍。解决方案：加盐打散热点Key、广播小表避免Shuffle、Spark AQE自动处理倾斜Partition。数据倾斜是Shuffle优化中最难的部分。

总结

Shuffle面试讲清楚三个层面：① Map端过程（排序→溢写→Combiner→合并）；② Reduce端过程（拉取→归并→聚合→输出）；③ 优化方法（Combiner减少数据量、列裁剪不SELECT *、压缩传输、调大缓冲区、合理设置Reduce数）。Shuffle是所有大数据框架（MR/Spark/Flink）数据交换的核心机制——理解了MR的Shuffle，也就理解了Spark和Flink的数据交换原理。能讲出SELECT *改列裁剪Shuffle减少45%的实际案例，面试官会很满意。

下一篇聊100TB的Hive表跑了6小时——大规模数据处理的调优思路和实战经验。

你们做过什么Shuffle优化？效果怎么样？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长