PySpark 3.0中缓存后小DataFrame写入Parquet时Shuffle Write异常暴增问题咨询

核心问题：缓存未正确生效，导致写入时重复计算全量数据链 #

你遇到的Shuffle Write暴增，本质是缓存没有真正落地生效，导致执行write操作时，Spark重新计算了df3的整个依赖链路——包括之前crossjoin生成20亿行的步骤，这才会产生300GB的Shuffle数据。

为什么缓存没生效？因为你搞反了操作顺序：

df3.count() # 触发全量计算，但结果未被缓存
df3=df3.cache() # 仅标记要缓存，无实际持久化操作

cache()是转换操作，必须搭配后续的Action（比如count()、collect()）才能触发缓存写入。你先执行count()计算完df3，再调用cache()，此时缓存标记是加在已计算的DataFrame上，但没有新的Action触发持久化，所以write执行时，Spark会重新跑一遍全量计算流程，自然又产生了海量Shuffle。

针对性解决方案 #

1. 修正缓存顺序，确保缓存生效

把cache()放在count()之前，让count()同时触发计算和缓存持久化：

df3 = df3.cache() # 先标记要缓存
df3.count() # 触发计算，同时将结果持久化到缓存
df3.write.partitionBy('date').parquet('some_file_name')

这样后续的write操作会直接读取缓存数据，不会重复计算crossjoin的海量数据。

2. 优化小数据集的分区数

你的spark.sql.shuffle.partitions=200是针对大计算设置的，但过滤后的df3只有400k行，200个分区会产生大量小文件，还会增加Shuffle开销。建议提前调整分区数：

• 用coalesce（不触发Shuffle）快速减少分区：

  df3 = df3.cache()
  df3 = df3.coalesce(10) # 可根据date的不同取值数量调整，比如date有20种就设为20
  df3.count()
  

• 或者在写入时直接指定分区数：

  df3.write.partitionBy('date').option("numPartitions", 10).parquet('some_file_name')
  

3. 提前过滤，大幅减少中间计算量

既然最终只需要400k行数据，建议把过滤条件提前到crossjoin之后立刻执行，避免对20亿行数据跑UDF：

df3 = df1.crossjoin(F.broadcast(df2)) \
         .filter(你的过滤条件) \ # 先过滤出目标数据，再执行UDF
         .withColumn(..., 你的UDF操作)
df3.cache()
df3.count()

这样整个计算链的数据量会从20亿行直接降到400k行，后续缓存和写入的效率会大幅提升。

4. 临时调整Shuffle分区数

针对小数据集，临时降低spark.sql.shuffle.partitions的值，避免不必要的Shuffle开销：

# 用SQL命令设置
spark.sql("SET spark.sql.shuffle.partitions=10")
# 或者用API设置
spark.conf.set("spark.sql.shuffle.partitions", 10)

df3.write.partitionBy('date').parquet('some_file_name')

额外排查点 #

• 检查缓存状态：执行df3.storageLevel查看缓存级别（默认是MEMORY_AND_DISK），确保集群有足够内存/磁盘存放缓存数据；
• 查看Spark UI：在Storage标签页确认df3是否真的被缓存，以及缓存大小是否符合预期（200MB左右）；在Jobs标签页查看write任务的Stage，确认是否有重复计算crossjoin的步骤。

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