减少分区数量时，为什么spark数据帧重新分区比合并快？

Question

我有一个包含100个分区的df，在保存到HDFS之前，我想减少分区的数量，因为拼花文件太小了（

df.coalesce(3).write.mode("append").parquet(OUTPUT_LOC)

它可以工作，但将过程从每个文件 2-3 秒减慢到每个文件 10-20 秒。当我尝试重新分区时：

df.repartition(3).write.mode("append").parquet(OUTPUT_LOC)

这个过程一点也不慢，每个文件2-3秒。

为什么？在减少分区数量时，合并不应该总是更快，因为它避免了完全洗牌吗？

背景：

我将文件从本地存储导入spark集群，并将生成的数据帧保存为拼花文件。每个文件大约100-200MB。文件位于“spark driver”机器上，我在客户端部署模式下运行spark submit。我正在驱动器中逐个读取文件：

data = read_lines(file_name)
rdd = sc.parallelize(data,100)
rdd2 = rdd.flatMap(lambda j: myfunc(j))
df = rdd2.toDF(mySchema)
df.repartition(3).write.mode("append").parquet(OUTPUT_LOC)

Spark版本是3.1.1

Spark/HDFS集群有5个工作人员，8CPU，32GB RAM

每个执行器有4个内核和15GB RAM，总共有10个执行器。

编辑：

当我使用coalesce（1）时，我得到spark.rpc.message.max大小限制违反错误，但当我使用重新分区（1）时没有。这可能是一个线索吗？

附加DAG可视化..看起来WholeStageCodegen部分在合并Dag上花费了太长时间？

Answer 1

如果您的数据分布不均匀，有时会发生这种情况，当您合并数据时，它会尝试通过合并小分区来减少分区，以减少完全混排，但其中一个分区中仍可能存在一些数据倾斜，并且该分区将占用大部分时间。

当您重新分区时，数据几乎均匀地分布在所有分区上，因为它会进行完全随机播放，并且所有任务几乎可以同时完成。

您可以使用spark UI来查看为什么当您合并任务时会发生什么，以及您是否看到任何单个任务长时间运行。