如何高效执行Spark数据加载转换并解决shuffle运行异常

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 0报错的核心原因是Shuffle阶段Map输出元数据丢失，和业务逻辑无关——14万条数据规模、无嵌套循环的计算远未达到集群算力瓶颈，问题全部来自不合理的默认/自定义配置导致Executor异常退出、Shuffle服务资源不足。

• 首先纠正不合理的内存配置
  单Worker节点总内存仅64GB，当前配置给Executor分配64GB内存完全没有给操作系统、节点守护进程、Shuffle服务预留内存，会触发系统OOM Killer直接杀掉占用内存过高的Executor进程，对应Executor生成的Shuffle文件和元数据会直接丢失，是本次报错的核心原因。调整规则如下：
  • 单节点8核CPU，按每个Executor分配4核计算，单节点可运行2个Executor，每个Executor分配30~35GB内存即可，单节点预留至少10GB内存给OS、NodeManager、外部Shuffle服务使用
  • Driver内存无需设置64GB，14万条数据规模下Driver分配8~16GB完全足够，过高的Driver内存会挤占Worker节点运行资源
  • 移除spark.driver.maxResultSize=0的无限制配置，改为spark.driver.maxResultSize=4g，避免结果集异常撑爆Driver内存
  • 关闭当前开启的推测执行，设置spark.speculation=false，资源紧张时推测执行会启动重复任务抢占资源，反而会导致正常运行的任务被kill，丢失Shuffle元数据，你当前数据规模不存在明显慢节点问题，不需要开启该特性
• 补全Shuffle稳定性配置
  • 开启外部Shuffle服务：配置spark.shuffle.service.enabled=true，由节点上独立的Shuffle服务托管Shuffle输出文件，即使Executor被回收，后续任务依然可以读取到已生成的Shuffle数据
  • 调整Shuffle分区数：将spark.sql.shuffle.partitions从默认的200调整为72_{144（匹配集群总CPU核数72，按1}2倍核数设置即可），分区过多会产生大量小Shuffle文件提升元数据读取压力，分区过少会导致单分区数据量过大触发OOM
  • 增加Shuffle拉取重试容错：配置spark.shuffle.io.maxRetries=10、spark.shuffle.io.retryWait=30s，避免网络瞬时波动导致元数据拉取失败直接报错

• 优先优化Join逻辑，消除不必要的Shuffle
  你的场景中大部分关联表是和主表主键关联的列表数据，数据量远小于主表，直接使用广播Join替代默认的Sort Merge Join，完全跳过Shuffle阶段：
  1. 将spark.sql.autoBroadcastJoinThreshold从默认的10MB调整为1GB，让Spark自动识别可广播的小表
  2. 对于明确数据量不大的关联表，代码中可手动加广播提示，示例：

  import org.apache.spark.sql.functions.broadcast
  val result = mainDs
    .join(broadcast(listTable1), Seq("primary_key"), "left")
    .join(broadcast(listTable2), Seq("primary_key"), "left")
    // 其余关联表同理处理
  

• 优化集合类型列生成逻辑
  不要等5张表全Join完成后再聚合生成集合列，先对每个存储列表数据的关联表，提前按关联主键做聚合，用collect_list/collect_set生成对应集合列，聚合后每个主键仅对应1行数据，再和主表做关联，避免Join阶段产生数据膨胀，大幅减少参与计算的数据量。
• 算子逻辑优化
  • 简单字段计算优先使用Spark SQL内置函数实现，不要在map()中用自定义lambda实现相同逻辑，Catalyst优化器会对内置函数做字节码级优化，性能比自定义lambda高2~10倍
  • 如果必须使用自定义lambda，优先用mapPartitions替代普通map，把正则编译、日期格式化这类可复用的对象初始化放在分区级逻辑里，避免每条记录重复创建对象增加GC压力
• 数据库读取优化
  从数据库抽取数据时，对每张表按主键配置分区读取参数，设置numPartitions等于集群总Executor核数，按主键范围切分读取任务，实现全并行拉取，避免单线程读取全表导致的拉取慢、数据倾斜问题。

内容的提问来源于stack exchange，提问作者Juan Daniel Cortez Rojas