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.cores8这些Task共享JVM堆内存。如果Map Task把大量中间结果全塞进内存等Reduce来拉极易触发GC甚至OOM而Reduce Task一边拉数据一边做聚合也需要自己的内存空间。磁盘在这里充当了天然的“缓冲区隔离带”Map Task写完就释放内存Reduce Task按需读取双方内存压力解耦。ShuffleWriteMetrics里专门统计shuffleWriteTime和shuffleWriteBytes就是为了量化这种“用磁盘换内存”的代价。网络拓扑不可控生产环境集群节点间网络质量参差不齐。假设Map Task在节点AReduce 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. BypassMergeSortShuffleWriterSpark 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写入策略的开关它的判断条件非常具体Shuffle依赖不能有聚合操作即handle.dependency必须是ShuffleDependency且aggregator.isEmpty为true。换句话说只有mapToPair(...).reduceByKey(...)这种带聚合的才走Sort路径而mapToPair(...).groupByKey()或纯repartition()这种不聚合的才可能走Bypass。分区数不能太多conf.get(config.SHUFFLE_SORT_BYPASS_MERGE_THRESHOLD) 200默认值。如果rdd.partitions.length 200认为分区数少直接为每个partition创建一个独立文件成本可控否则分区太多比如上万生成上万个文件会导致文件系统inode耗尽必须走Sort路径做合并。序列化器必须支持重定位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.enabledtrue打开后Spark SQL的自适应查询优化AQE会动态合并小分区导致原本1000个分区被合并成50个。结果shouldBypassMergeSort()判断分区数50200自动切到Bypass路径。但上游数据倾斜严重某个partition数据量是其他partition的100倍Bypass为它单独生成一个2GB的文件而其他49个partition文件都不到1MB。下游Reduce Task拉取这个2GB文件时单线程读取解压缩耗时飙升拖垮整个Stage。解决方案不是关AQE而是显式设置spark.shuffle.sort.bypassMergeThreshold10强制走Sort路径让大partition的数据在merge阶段被切分成多个有序segmentReduce端可并行拉取。这说明路径选择不是非黑即白要结合数据分布特征做针对性干预。2.3 磁盘布局的底层约定从ShuffleBlockResolver到文件系统的真实映射无论走哪条路径最终生成的文件都必须遵循一套严格的命名和目录规则否则Reduce端根本找不到该读哪个文件。这套规则由ShuffleBlockResolver接口定义其实现类IndexShuffleBlockResolver负责具体落地。我们来看一个真实例子假设shuffleId10,mapId5,reduceId3那么该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只需打开.index文件seek(8*1)定位到第2个8字节readLong()得到offset4660再seek(4660)跳到.data文件对应位置按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.compresstrue默认true则fileOutputStream会被LZ4CompressorStream包装stream: SerializationStream序列化流由serializer.serializeStream()创建。当sorter.writeSortedFile()开始写入时它会调用writer.open()此时发生fileOutputStream new FileOutputStream(file, append false)—— 创建新文件覆盖模式如果启用压缩compressStream new LZ4CompressorStream(fileOutputStream)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())。这种设计极大减少了系统调用次数但代价是增加了内存占用。实测数据当bufferSize32KB时每秒系统调用write()约1200次当bufferSize128KB时降到约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初始化时它会构建fetchRequests: mutable.Queue[FetchRequest]每个FetchRequest包含目标host:port、要拉取的blockId列表如shuffle_10_5_3、以及该请求的优先级启动一个后台线程fetchContinuous()持续从队列取FetchRequest通过TransportClient发起Netty请求拉取到的数据块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。而maxRetries3意味着单个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.enabledtrue且spark.shuffle.service.cleanup.interval300s5分钟由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 --nameseqwrite --ioenginepsync --rwwrite --bs128k --size10G --runtime300 --time_based --filename/data1/testfile # 测试随机读模拟shuffle read因为index寻址是随机的 fio --namerandread --ioenginelibaio --rwrandread --bs4k --size10G --runtime300 --time_based --filename/data1/testfile --iodepth64 # 测试混合读写模拟Bypass路径的并发读写 fio --namemixed --ioenginelibaio --rwrandrw --rwmixread70 --bs8k --size10G --runtime300 --time_based --filename/data1/testfile --iodepth32重点关注三个指标seqwrite的bw带宽应≥500MB/sNVMe SSD或≥150MB/sSATA SSDrandread的iops每秒IO次数应≥50KNVMe或≥10KSATAmixed的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对比模板Ashuffle 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.maxSizeInFlight48MB96MB~256MB增大可提升吞吐但需确保Executor内存足够spark.executor.memory * 0.6 256MBspark.reducer.maxReqsInFlight510~20增加并发拉取请求数尤其适合多磁盘环境但过多会压垮网络spark.reducer.maxBlocksInFlightPerAddress100200~500单IP最大并发block数防止单节点被打爆spark.shuffle.detectCorrupttruetrue必须开启自动校验block CRC避免静默数据损坏一个典型配置组合spark.reducer.maxSizeInFlight 134217728 # 128MB spark.reducer.maxReqsInFlight 15 spark.reducer.maxBlocksInFlightPerAddress 300 spark.shuffle.detectCorrupt true注意maxReqsInFlight和maxBlocksInFlightPerAddress要配合使用。比如集群有100个Executor每个Executor有4块SSD那么maxReqsInFlight15意味着单个Reduce Task最多同时向15个不同Executor发请求而maxBlocksInFlightPerAddress300意味着向单个Executor最多并发拉300个block。这样既利用了多磁盘并行又不会让单个Executor的网络连接数爆炸。5. 常见问题与排查技巧实录来自生产环境的12个真实故障案例5.1 故障速查表症状、根因、解决方案症状可能根因快速验证命令解决方案ShuffleBlockFetcherIterator: Failed to fetch blockConnection refusedExternalShuffleService未启动或端口不通telnet executor_host 7337检查spark.shuffle.service.enabledtrue确认服务进程存活No space left on device但df -h显示磁盘充足inode耗尽大量小文件df -i清理/tmp/spark-*调小bypassMergeThreshold启用spark.sql.adaptive.coalescePartitions.enabledShuffle read time异常高fetchWaitTime占比80%网络带宽打满或磁盘IO wait高sar -n DEV 1 5网卡iostat -x 1 5磁盘增加maxReqsInFlight检查交换机QoS升级网卡驱动java.lang.OutOfMemoryError: Java heap space在ShuffleBlockFetcherIteratormaxSizeInFlight过大或数据倾斜jstat -gc pid观察老年代增长降低maxSizeInFlight用salting缓解倾斜.index文件存在但.data文件缺失Map Task写入中途崩溃.index未同步更新ls -la /path/to/shuffle/*/5/*.index对比.data重启失败Task检查spark.shuffle.service.cleanup.interval是否过短BlockId not found错误ExternalShuffleService未配置spark.shuffle.service.portps aux | grep shuffle查看服务启动参数显式设置spark.shuffle.service.port7337shuffle write time波动剧烈忽高忽低磁盘IO被其他进程抢占iotop -p executor_pid将spark.local.dir绑定到独立SSD禁用cron等定时任务fetchFailed计数持续增长网络丢包或防火墙拦截mtr --report executor_host调大maxRetries检查防火墙规则Shuffle write bytes远小于Shuffle read bytesMap端数据被压缩Reduce端解压后膨胀spark.shuffle.spill.compresstrue监控Shuffle Write Metrics的shuffleWriteTime和shuffleWriteBytes比值若5说明压缩收益高java.io.IOException: Stream closedDiskBlockObjectWriter被重复close()查看Task日志中writer.close()调用栈检查自定义ShuffleManager实现确保stop()只调用一次shuffle read records为0但shuffle read bytes很大数据被过滤或mapPartitions中return Nilspark.sql.adaptive.enabledfalse临时关闭AQE在mapPartitions中添加println(spartition size: ${iter.size})调试Web UI显示Shuffle Write Size但无Shuffle Read SizeReduce Task未启动或Stage卡在调度yarn logs -applicationId app_id检查spark.scheduler.modeFAIR是否配置公平调度池资源是否被抢占5.2 独家避坑技巧那些文档里不会写的细节技巧1用spark.ui.retainedStages5保活历史Stage UI默认Spark只保留最近10个Stage的UI信息。当排查shuffle问题时往往需要对比多个Stage的metrics。设置spark.ui.retainedStages50并在spark.ui.retainedJobs50这样Web UI能长期看到历史shuffle行为方便横向对比