Pyspark计算TF-IDF时出现MetadataFetchFailedException报错如何解决

1 报错根本原因说明 #

tf和idf的count值差异不是引发报错的原因，普通join操作本身就是为了处理不同基数的KV关联场景，不会触发该异常。
你遇到的MetadataFetchFailedException: Missing an output location for shuffle 4属于Spark shuffle阶段的典型错误，根因是reduce任务拉取shuffle数据时，找不到对应map任务的输出元数据，一般由以下情况触发：

• executor内存不足触发Full GC，长时间停顿导致节点被标记为失联，对应map任务的输出元数据被清空
• shuffle服务异常、网络波动导致map输出的元数据同步失败
  你已经调整了重试和超时参数但问题依然存在，核心是shuffle阶段的内存压力没有得到缓解。

2 代码优化方案 #

不需要执行高代价的分区校验操作，你的场景可以通过更高效的join方式彻底规避不必要的shuffle：
idf是按单词维度去重统计的结果，数据量远小于按「文档+单词」维度统计的tf，完全可以将idf转为本地集合后广播到所有executor，用广播join代替普通shuffle join，从根源上消除触发报错的shuffle操作，代码修改参考：

# 先将小体量的idf RDD收集到Driver端，转为字典后广播
idf_dict = dict(idf.collect())
broadcast_idf = spark.sparkContext.broadcast(idf_dict)

# tf端直接用广播变量做关联，不需要触发shuffle
tfidf = tf.map(lambda x: (x[0], (x[1], broadcast_idf.value.get(x[0]))))
# 业务允许的前提下可新增过滤逻辑，去掉无对应idf的无效数据
# tfidf = tfidf.filter(lambda x: x[1][1] is not None)

如果确实需要保留普通join逻辑，也可以提前对两个RDD设置合理的分区数，减少单分区的数据量：

# 调整shuffle分区数为核心数的2~3倍，默认200的分区数对你的小数据集来说反而有调度开销
spark.conf.set("spark.sql.shuffle.partitions", 12)
tf = tf.repartition(12)
idf = idf.repartition(12)
tfidf = tf.join(idf)

3 集群配置优化建议 #

你的数据集只有569MB，属于极小体量，当前配置的问题是内存分配不合理，且总并行度过低，反而容易引发内存碎片和GC，推荐配置如下（单作业专属资源）：

• spark.executor.instances：2~3个就足够
• spark.executor.cores：每个executor分配3~4核，充分利用CPU的同时避免单executor任务过多导致内存争抢
• spark.executor.memory：每个executor分配3G堆内存
• spark.executor.memoryOverhead：每个executor分配1G堆外内存
• 额外新增配置：
  • spark.shuffle.memoryFraction：调整为0.3，给shuffle阶段预留更多内存，避免shuffle数据频繁溢写到磁盘
  • spark.shuffle.sort.bypassMergeThreshold：调整为1000，小分区shuffle时绕过排序逻辑，减少CPU和内存开销

4 额外排查点 #

如果修改后依然报错，可以通过Spark UI的Executors标签页查看对应报错executor的GC时间和日志，确认是否存在OOM被kill的情况，如果是集群共享资源的场景，可以适当调高spark.executor.instances降低单executor的负载。

内容的提问来源于stack exchange，提问作者user2829319