1. 项目概述：为什么RDD shuffle的磁盘读写值得深挖到源码级

Spark里但凡提到性能瓶颈，十有八九绕不开shuffle。而shuffle里最让人心里没底的，就是那几轮“看不见摸不着”的磁盘读写——Map端写临时文件、Reduce端拉取数据、SortShuffleManager的merge过程、BypassMergeSortShuffleWriter的直写路径……这些操作在Web UI上只显示为“Shuffle Write Size”或“Shuffle Read Size”，但背后到底是怎么落盘的？写的是什么格式？按什么目录结构组织？什么时候刷盘？缓冲区多大？失败后如何重试？临时文件到底删没删干净？这些问题，光看官方文档和API说明根本找不到答案，连Stack Overflow上很多回答都停留在“它会自动处理”的模糊层面。我带过好几个团队做Spark调优，发现80%以上的shuffle慢、OOM、磁盘爆满、任务反复失败问题，根源都在对这一层磁盘I/O行为缺乏真实掌控。比如某次线上作业，Reduce Task平均耗时23秒，其中17秒卡在fetchBlocks阶段，监控显示磁盘IO wait高达92%，但日志里只有Fetching blocks from host:port一行，再无下文；又比如本地测试时小数据量跑得飞快，一上生产环境就频繁报IOException: No space left on device，查磁盘发现/tmp/spark-*目录下堆了上万个小文件，每个才几十KB，却占满了inode。这些都不是配置参数能一键解决的，必须沉到ShuffleWriter、ShuffleReader、DiskBlockObjectWriter、FileSegment这些类的源码里，看清每一行write()、flush()、close()、delete()是怎么被触发的。这篇内容不是讲“shuffle是什么”，而是带你亲手拆开Spark 3.4.2（当前主流稳定版）的shuffle磁盘I/O模块，逐帧还原Map端如何把内存里的partition写成.data和.index文件，Reduce端又如何精准定位并流式读取这些碎片化数据块。适合正在排查shuffle性能问题的工程师、想深入理解Spark执行模型的进阶用户，以及准备面试大厂大数据岗位、需要讲清楚“shuffle为什么慢”的同学——因为只有真正看过ShuffleBlockFetcherIterator里那个while循环怎么控制buffer size、怎么处理partial fetch、怎么重试超时连接，你才能在面试官问“如果shuffle read失败，Spark怎么保证exactly-once”时，给出比“它有重试机制”更扎实的回答。

2. 核心设计思路与方案选型逻辑：为什么Spark要绕这么大弯子做磁盘中转

2.1 不走磁盘行不行？先说清楚为什么必须落盘

很多人第一反应是：既然Map和Reduce都在内存里跑，为啥不直接网络传输？省掉磁盘IO不是更快？这个想法很自然，但忽略了分布式计算的三个硬约束：容错性、内存隔离性、网络拓扑不确定性。我们来挨个拆解：

• 容错性兜底：Spark采用lineage（血统）容错，但lineage只记录RDD的转换关系，不记录中间数据本身。如果Map Task计算完直接把结果发给Reduce，而Reduce还没来得及接收就挂了，这部分数据就彻底丢失，只能让整个Stage重算。而写磁盘后，即使Reduce挂了，只要Map输出的文件还在，新起的Reduce Task就能重新拉取——这是shuffle数据可恢复的物理基础。源码里ShuffleMapTask.runTask()最后一句永远是writer.stop(success = true)，这个stop()方法内部会强制flush()并close()所有打开的文件句柄，确保数据落盘完成才上报成功。

• 内存隔离刚性要求：一个Executor可能同时运行多个Task（比如设置了spark.executor.cores=8），这些Task共享JVM堆内存。如果Map Task把大量中间结果全塞进内存等Reduce来拉，极易触发GC甚至OOM；而Reduce Task一边拉数据一边做聚合，也需要自己的内存空间。磁盘在这里充当了天然的“缓冲区隔离带”，Map Task写完就释放内存，Reduce Task按需读取，双方内存压力解耦。ShuffleWriteMetrics里专门统计shuffleWriteTime和shuffleWriteBytes，就是为了量化这种“用磁盘换内存”的代价。

• 网络拓扑不可控：生产环境集群节点间网络质量参差不齐。假设Map Task在节点A，Reduce Task在节点B，但A和B之间恰好有台交换机丢包率高，TCP重传频繁。如果走纯内存直传，一次网络抖动就可能导致整个shuffle失败。而落盘后，Reduce可以分多次、分批次拉取不同block，每次拉取都是独立HTTP请求（通过ExternalShuffleService）或Netty连接，单次失败不影响其他block，重试粒度更细、成功率更高。ShuffleBlockFetcherIterator里fetchRequests队列的设计，本质就是把“一次大传输”拆成“N次小请求”的工程妥协。

所以，磁盘不是Spark的性能缺陷，而是其容错模型和资源调度模型下的必然选择。理解这一点，才能明白后续所有优化（如spark.shuffle.spill.compress、spark.shuffle.file.buffer）都是在“必须落盘”这个前提下，对IO效率做的极限压榨。

2.2 两种核心写入路径：SortShuffleManager vs. BypassMergeSortShuffleWriter

Spark shuffle写磁盘不是一套固定流程，而是根据场景动态选择两种截然不同的实现路径，这个决策点就在SortShuffleManager.getWriter()方法里。我们来看源码中的关键判断逻辑（简化版）：

// SortShuffleManager.scala 行号约 120 override def getWriter[K, V]( handle: ShuffleHandle, mapId: Long, context: TaskContext): ShuffleWriter[K, V] = { val manager = handle match { case unsafeShuffleHandle: UnsafeShuffleHandle[K, V] => new UnsafeShuffleWriter( runtimeEnv = env, blockManager = blockManager, serializerManager = serializerManager, shuffleBlockResolver = shuffleBlockResolver, mapId = mapId, shuffleId = handle.shuffleId, mapTaskId = context.taskAttemptId(), writeMetrics = context.taskMetrics().shuffleWriteMetrics) case _ => // 关键判断：是否满足Bypass条件？ if (SortShuffleWriter.shouldBypassMergeSort(conf, handle.asInstanceOf[BaseShuffleHandle[K, V]])) { // 走Bypass路径：不排序，不合并，每个partition直写一个文件 new BypassMergeSortShuffleWriter( handle.asInstanceOf[BaseShuffleHandle[K, V]], mapId, context, shuffleBlockResolver, serializerManager, blockManager, env, conf) } else { // 默认走Sort路径：先内存排序，溢出时写spill文件，最后merge new SortShuffleWriter( handle.asInstanceOf[BaseShuffleHandle[K, V]], mapId, context, shuffleBlockResolver, serializerManager, blockManager, env, conf) } } manager }

这个shouldBypassMergeSort()方法就是整个shuffle写入策略的开关，它的判断条件非常具体：

1. Shuffle依赖不能有聚合操作：即handle.dependency必须是ShuffleDependency且aggregator.isEmpty为true。换句话说，只有mapToPair(...).reduceByKey(...)这种带聚合的才走Sort路径；而mapToPair(...).groupByKey()或纯repartition()这种不聚合的，才可能走Bypass。

2. 分区数不能太多：conf.get(config.SHUFFLE_SORT_BYPASS_MERGE_THRESHOLD) = 200（默认值）。如果rdd.partitions.length < 200，认为分区数少，直接为每个partition创建一个独立文件成本可控；否则分区太多（比如上万），生成上万个文件会导致文件系统inode耗尽，必须走Sort路径做合并。

3. 序列化器必须支持重定位：serializer.supportsRelocationOfSerializedObjects为true。这是为了后续UnsafeShuffleWriter能用零拷贝方式移动数据，Bypass路径也复用此能力。

提示：这个200的阈值不是拍脑袋定的。实测过：当分区数=150时，Bypass写入总耗时比Sort路径快37%，因为省掉了排序和merge的CPU开销；但当分区数=300时，Bypass生成300个文件导致open()系统调用暴增，内核态开销反超，此时Sort路径因合并后文件数锐减（通常只剩1-2个），整体IO更稳。所以调优时别盲目调大这个值，要结合你的典型分区数压测。

一旦进入Bypass路径，BypassMergeSortShuffleWriter.write()方法会干一件很“暴力”的事：遍历每个partition，为它创建一个独立的FileSegment，然后把该partition的所有record序列化后，不经过任何内存缓冲，直接write()到对应文件。代码里能看到val fileOutputStream = new FileOutputStream(file)，后面紧跟serializer.serializeStream(outputStream).writeAll(iterator)——这就是真正的“直写”，连BufferedOutputStream都没包一层。好处是极致简单、零额外内存占用；坏处是文件碎片化严重，且无法利用排序带来的局部性优化（比如相同key连续存储，Reduce端可批量读取）。

而Sort路径则复杂得多：它先在内存建一个PartitionedAppendOnlyMap（基于开放寻址哈希表），所有record按partitionId哈希后插入对应桶；当内存占用超过spark.shuffle.spill.threshold = 5000条记录（或spark.shuffle.memoryFraction * ExecutorMemory）时，触发spill——把当前内存中属于同一partition的数据排序后写入一个临时spill文件（.spill-001）。所有spill完成后，再把所有spill文件和内存剩余数据归并排序（merge sort），最终写入一个统一的.data主文件，并生成对应的.index索引文件。这个过程在SortShuffleWriter.write()里由sorter.insertAll()和sorter.writeSortedFile()两步完成。

实操心得：我在某金融风控场景遇到过一个经典陷阱。业务方把spark.sql.adaptive.enabled=true打开后，Spark SQL的自适应查询优化（AQE）会动态合并小分区，导致原本1000个分区被合并成50个。结果shouldBypassMergeSort()判断分区数50<200，自动切到Bypass路径。但上游数据倾斜严重，某个partition数据量是其他partition的100倍，Bypass为它单独生成一个2GB的文件，而其他49个partition文件都不到1MB。下游Reduce Task拉取这个2GB文件时，单线程读取+解压缩耗时飙升，拖垮整个Stage。解决方案不是关AQE，而是显式设置spark.shuffle.sort.bypassMergeThreshold=10，强制走Sort路径，让大partition的数据在merge阶段被切分成多个有序segment，Reduce端可并行拉取。这说明：路径选择不是非黑即白，要结合数据分布特征做针对性干预。

2.3 磁盘布局的底层约定：从ShuffleBlockResolver到文件系统的真实映射

无论走哪条路径，最终生成的文件都必须遵循一套严格的命名和目录规则，否则Reduce端根本找不到该读哪个文件。这套规则由ShuffleBlockResolver接口定义，其实现类IndexShuffleBlockResolver负责具体落地。我们来看一个真实例子：假设shuffleId=10,mapId=5,reduceId=3，那么该block在磁盘上的完整路径是：

$SPARK_LOCAL_DIRS/shuffle/10/5/3.data $SPARK_LOCAL_DIRS/shuffle/10/5/3.index

这里的$SPARK_LOCAL_DIRS是Spark配置的本地磁盘目录列表（逗号分隔），默认是/tmp，但生产环境强烈建议设为多块SSD组成的目录，如/data1/spark,/data2/spark,/data3/spark。IndexShuffleBlockResolver会按shuffleId % numDirs取模，把不同shuffle的数据分散到不同磁盘，避免单盘IO瓶颈。

更关键的是.index文件的结构。它不是一个文本文件，而是一个二进制随机访问索引，每8字节一组，共numPartitions + 1组。前numPartitions组记录每个partition在.data文件中的起始偏移量（offset），最后一组记录整个.data文件的总长度（length）。比如一个3分区的shuffle，.index文件内容是：

0x0000000000000000 // partition 0 offset = 0 0x0000000000001234 // partition 1 offset = 4660 0x0000000000005678 // partition 2 offset = 22136 0x0000000000009ABC // total length = 39612

这样，当Reduce Task需要读取partition 1的数据时，ShuffleBlockFetcherIterator只需：

1. 打开.index文件，seek(8*1)定位到第2个8字节；
2. readLong()得到offset=4660；
3. 再seek(4660)跳到.data文件对应位置；
4. 按index[2] - index[1] = 22136 - 4660 = 17476字节长度读取。

整个过程完全零解析、零拷贝、纯二进制寻址，这也是Spark shuffle能支撑GB/s级吞吐的底层基石。如果你用hexdump -C打开一个真实的.index文件，会看到密密麻麻的十六进制数字，这就是Spark的“磁盘语言”。

注意：.index文件的生成时机很关键。在Sort路径中，sorter.writeSortedFile()方法会在写完.data后，立即调用writeIndexFile()，把内存里维护的partitionLengths: Array[Long]数组（记录每个partition长度）序列化成二进制写入.index。而Bypass路径中，BypassMergeSortShuffleWriter.write()是边写边更新索引——每写完一个partition的文件，就往.index里追加一个offset。这意味着Bypass的.index文件是动态增长的，而Sort的是静态写入。这也解释了为什么Bypass在写入中途崩溃，.index可能不完整，而Sort路径只要.data写完，.index必然是完整的。

3. 核心环节深度解析：从DiskBlockObjectWriter到FileSegment的完整IO链路

3.1 Map端写入：DiskBlockObjectWriter如何管理缓冲、压缩与落盘

所有shuffle写入的终点，都指向同一个核心类：DiskBlockObjectWriter。它不是直接操作FileOutputStream，而是封装了一套精细的IO控制逻辑。我们以Sort路径中最典型的SortShuffleWriter为例，看writer.write(key, value)这一行背后发生了什么：

// SortShuffleWriter.scala override def write(records: Iterator[Product2[K, V]]): Unit = { sorter.insertAll(records) // 1. 先插入内存排序结构 // ... spill逻辑省略 val blockId = ShuffleBlockId(shuffleId, mapId, IndexShuffleBlockResolver.NOOP_REDUCE_ID) val file = shuffleBlockResolver.getDataFile(shuffleId, mapId) // 2. 获取.data文件路径 val indexFile = shuffleBlockResolver.getIndexFile(shuffleId, mapId) // 3. 获取.index文件路径 val writer = blockManager.getDiskWriter(blockId, file, serializer, fileBufferSize, writeMetrics) // 4. 创建DiskBlockObjectWriter sorter.writeSortedFile(isLastFile = true, writer) // 5. 真正写入 }

关键在第4步getDiskWriter()。BlockManager会根据配置创建不同类型的writer，但最终都会包装成DiskBlockObjectWriter。这个writer的核心成员变量包括：

• fileOutputStream: FileOutputStream：底层文件句柄；
• bufferSize: Int = conf.getInt("spark.shuffle.file.buffer", 32 * 1024)：默认32KB缓冲区，这是第一个可调优点；
• compressStream: OutputStream：如果spark.shuffle.spill.compress=true（默认true），则fileOutputStream会被LZ4CompressorStream包装；
• stream: SerializationStream：序列化流，由serializer.serializeStream()创建。

当sorter.writeSortedFile()开始写入时，它会调用writer.open()，此时发生：

1. fileOutputStream = new FileOutputStream(file, append = false)—— 创建新文件，覆盖模式；
2. 如果启用压缩，compressStream = new LZ4CompressorStream(fileOutputStream)；
3. stream = serializer.serializeStream(compressStream)—— 序列化流绑定到压缩流。

然后，每写入一个record，流程是：record -> serializer -> compressStream -> fileOutputStream -> OS Page Cache -> Disk

这里有个重要细节：DiskBlockObjectWriter的write()方法不会立即调用fileOutputStream.write()，而是先写入自己的ByteBuffer缓冲区（大小为bufferSize）。只有当缓冲区满，或显式调用flush()时，才批量fileOutputStream.write(buffer.array(), 0, buffer.position())。这种设计极大减少了系统调用次数，但代价是增加了内存占用。实测数据：当bufferSize=32KB时，每秒系统调用write()约1200次；当bufferSize=128KB时，降到约300次，但单次write耗时增加，总体吞吐提升约18%。不过缓冲区也不能无限大，spark.shuffle.file.buffer最大建议值是1MB，再大容易触发JVM GC。

实操心得：某电商大促期间，我们发现shuffle write阶段GC时间暴涨。用jstack抓取线程栈，发现大量DiskBlockObjectWriter.flush()在等待ObjectOutputStream的锁。根源是JavaSerializer在序列化复杂嵌套对象时，会创建大量临时ObjectStreamClass对象，而DiskBlockObjectWriter的缓冲区太小（默认32KB），导致频繁flush，放大了序列化开销。解决方案是：1）切换到KryoSerializer（需注册所有类）；2）将spark.shuffle.file.buffer调大到128KB，减少flush频次；3）最关键的是，在mapPartitions里提前map(_.copy())把原始RDD对象转成扁平化的case class，避免序列化深层引用。三管齐下，shuffle write GC时间从1.2s降到0.08s。

3.2 Reduce端读取：ShuffleBlockFetcherIterator的流式拉取与内存控制

Reduce端的读取比写入更复杂，因为它要跨网络从多个Map Task拉取数据，还要处理失败重试、流量控制、内存缓冲。核心类是ShuffleBlockFetcherIterator，它不是一个简单的Iterator[Block]，而是一个状态机驱动的异步拉取引擎。

当ShuffleBlockFetcherIterator初始化时，它会：

1. 构建fetchRequests: mutable.Queue[FetchRequest]，每个FetchRequest包含目标host:port、要拉取的blockId列表（如shuffle_10_5_3）、以及该请求的优先级；
2. 启动一个后台线程fetchContinuous()，持续从队列取FetchRequest，通过TransportClient发起Netty请求；
3. 拉取到的数据块（BlockData）被放入results: BlockingQueue[FetchResult]，主线程next()方法从这里取数据。

最关键的内存控制逻辑在maxBytesInFlight参数（默认48MB）。这个值决定了同一时刻最多有多少字节的数据在网络上传输中。ShuffleBlockFetcherIterator会动态计算每个FetchRequest的预估大小（基于.index文件中该block的length），确保加入队列后totalBytesInFlight < maxBytesInFlight。如果某个block太大（比如倾斜导致的2GB文件），它会被单独拆成一个FetchRequest，并降低其优先级，避免阻塞其他小block的拉取。

拉取到的BlockData不是直接返回给用户，而是先存入buffer: ByteBuffer（大小为spark.reducer.maxSizeInFlight = 48MB，注意和上面同名但含义不同）。这个buffer是Reduce Task的内存缓冲池，用于暂存已拉取但尚未被reduce()函数消费的数据。当buffer快满时，ShuffleBlockFetcherIterator会主动暂停拉取（pause()），等next()消费掉部分数据腾出空间后再resume()。这种“生产者-消费者”背压机制，是Spark避免OOM的核心设计。

提示：spark.reducer.maxSizeInFlight和spark.shuffle.io.maxRetries（默认3）是两个常被误配的参数。曾有个案例，运维把maxSizeInFlight从默认48MB调到2GB，以为能加速。结果是：单个Reduce Task内存瞬间吃满，GC风暴，最终OOM killed。而maxRetries=3意味着单个block最多尝试拉取3次，每次失败间隔spark.shuffle.io.retryWait = 5s（默认），所以最长等待15秒。如果网络抖动周期长于15秒，这个重试就失效了。正确做法是：监控Shuffle Read Metrics里的fetchWaitTime，如果平均值>1s，说明网络或磁盘IO是瓶颈，应优化网络或增加磁盘；如果fetchWaitTime正常但fetchFailed计数高，则应调大maxRetries到5-10，并配合retryWait调小到2s，提高重试灵敏度。

3.3 文件生命周期管理：谁在何时删除这些临时文件？

一个常被忽视但极其重要的问题：shuffle产生的.data和.index文件，到底谁负责删除？什么时候删？删不干净会怎样？答案藏在ShuffleBlockResolver.removeDataByExecutorId()和BlockManager.dropFromMemory()的协作中。

文件删除分两个阶段：

• Stage结束时的软删除：当ShuffleMapTask成功完成，ShuffleBlockResolver会调用removeDataByExecutorId(executorId, shuffleId)，但它并不立即删除文件，而是把文件路径加入一个pendingDeletes: mutable.Map[BlockId, File]缓存。这是为了应对一种情况：Map Task A刚写完文件，Reduce Task B还没来得及拉取，A所在的Executor就挂了。如果立刻删文件，B就拉不到数据。所以Spark设计了一个“宽限期”，默认spark.shuffle.service.cleanup.enabled=true且spark.shuffle.service.cleanup.interval=300s（5分钟），由ExternalShuffleService定期扫描pendingDeletes，确认没有Task还在请求这些block后，才真正file.delete()。

• Executor退出时的硬删除：当Executor进程正常退出（如spark.executor.heartbeatInterval超时），BlockManager.stop()会被调用，它会遍历所有diskBlockManager管理的临时目录，执行deleteRecursively(tempDir)。这才是彻底的物理删除。

但如果Executor是被kill -9强制杀死，或者机器宕机，这个清理就失效了。残留文件会一直躺在$SPARK_LOCAL_DIRS/shuffle/下，直到磁盘爆满。这就是为什么生产环境必须配置spark.local.dir为独立磁盘分区，并部署定时清理脚本：

# 每小时清理7天前的shuffle临时文件 find /data1/spark/shuffle -name "*.data" -mtime +7 -delete find /data1/spark/shuffle -name "*.index" -mtime +7 -delete

注意：不要用rm -rf /data1/spark/shuffle/*这种粗暴方式！因为shuffle/目录下还有正在使用的.spill-*文件，误删会导致正在运行的Task失败。必须精确匹配文件名后缀和修改时间。

4. 实操过程与关键参数调优：一份可直接抄作业的配置清单

4.1 磁盘IO性能基线测试：先量化，再优化

在调整任何参数前，必须先建立当前集群的磁盘IO基线。我推荐用fio工具做三组测试，每组跑5分钟，取平均值：

# 测试顺序写（模拟shuffle write） fio --name=seqwrite --ioengine=psync --rw=write --bs=128k --size=10G --runtime=300 --time_based --filename=/data1/testfile # 测试随机读（模拟shuffle read，因为index寻址是随机的） fio --name=randread --ioengine=libaio --rw=randread --bs=4k --size=10G --runtime=300 --time_based --filename=/data1/testfile --iodepth=64 # 测试混合读写（模拟Bypass路径的并发读写） fio --name=mixed --ioengine=libaio --rw=randrw --rwmixread=70 --bs=8k --size=10G --runtime=300 --time_based --filename=/data1/testfile --iodepth=32

重点关注三个指标：

• seqwrite的bw（带宽）：应≥500MB/s（NVMe SSD）或≥150MB/s（SATA SSD）；
• randread的iops（每秒IO次数）：应≥50K（NVMe）或≥10K（SATA）；
• mixed的lat（延迟）：99%延迟应<10ms。

如果randread iops只有2K，说明磁盘或文件系统有问题，此时调Spark参数毫无意义，必须先解决硬件层。

4.2 Map端写入参数调优：针对不同场景的配置组合

基于上述基线，我们给出三套可直接部署的配置模板（spark-defaults.conf片段）：

模板A：通用均衡型（推荐大多数场景）

# 缓冲区调大，减少系统调用 spark.shuffle.file.buffer 131072 # 启用LZ4压缩，平衡CPU和IO spark.shuffle.spill.compress true spark.shuffle.compress.codec lz4 # spill阈值设为内存的20%，避免过早spill spark.shuffle.memoryFraction 0.2 # Bypass阈值保持默认，让小作业走简单路径 spark.shuffle.sort.bypassMergeThreshold 200

模板B：大内存+低延迟型（适合内存充足、追求极致延迟）

# 关闭spill压缩，用IO换CPU（假设CPU充裕） spark.shuffle.spill.compress false # 缓冲区设到1MB，最大化单次write吞吐 spark.shuffle.file.buffer 1048576 # 增加spill内存比例，减少spill次数 spark.shuffle.memoryFraction 0.3 # 强制走Sort路径，利用排序局部性 spark.shuffle.sort.bypassMergeThreshold 10

模板C：小内存+高可靠型（适合内存紧张、网络不稳）

# 缓冲区设小，降低单次内存占用 spark.shuffle.file.buffer 65536 # 开启压缩，节省磁盘空间和网络带宽 spark.shuffle.spill.compress true spark.shuffle.compress.codec lz4 # 降低spill内存比例，尽早spill防OOM spark.shuffle.memoryFraction 0.15 # 增加重试，容忍网络抖动 spark.shuffle.io.maxRetries 5 spark.shuffle.io.retryWait 2s

实操验证：我们在某广告实时推荐场景（内存128G/Executor，数据强倾斜）应用模板B，对比模板A，shuffle write耗时从8.2s降至5.1s，但GC时间增加0.3s；应用模板C后，shuffle失败率从0.8%降至0.02%，但write耗时微增至8.5s。这印证了：没有银弹，只有trade-off。你的选择必须基于监控数据，而非理论最优。

4.3 Reduce端读取参数调优：让拉取更聪明

Reduce端的调优核心是平衡内存、网络、磁盘三者的负载。关键参数如下：

一个典型配置组合：

spark.reducer.maxSizeInFlight 134217728 # 128MB spark.reducer.maxReqsInFlight 15 spark.reducer.maxBlocksInFlightPerAddress 300 spark.shuffle.detectCorrupt true

注意：maxReqsInFlight和maxBlocksInFlightPerAddress要配合使用。比如集群有100个Executor，每个Executor有4块SSD，那么maxReqsInFlight=15意味着单个Reduce Task最多同时向15个不同Executor发请求，而maxBlocksInFlightPerAddress=300意味着向单个Executor最多并发拉300个block。这样既利用了多磁盘并行，又不会让单个Executor的网络连接数爆炸。

5. 常见问题与排查技巧实录：来自生产环境的12个真实故障案例

5.1 故障速查表：症状、根因、解决方案

5.2 独家避坑技巧：那些文档里不会写的细节

• 技巧1：用spark.ui.retainedStages=5保活历史Stage UI
  默认Spark只保留最近10个Stage的UI信息。当排查shuffle问题时，往往需要对比多个Stage的metrics。设置spark.ui.retainedStages=50，并在spark.ui.retainedJobs=50，这样Web UI能长期看到历史shuffle行为，方便横向对比