Spark两种分区Join方案的性能差异原因及选型咨询

背景 #

我有两个按partition_column列分区的DataFrame（df1和df2），在执行Join操作时观察到两种方案存在明显性能差异：

方案1：利用Spark并行能力进行Join #

val writeDF = df1.as("df1").join(df2.as("df2"),
            col("df1.partition_column").equalTo("df2.partition_column")
            and col("df1.unique_id").equalTo(col("df2.unique_id"))).select("df1.*")
writeDF.write.mode(SaveMode.Overwrite).partitionBy("partition_column").parquet("some location")

执行计划显示该方案使用了分区过滤，但速度慢于方案2。

方案2：分组提交分区进行Join #

val partitions = df2.select("partition_column").distinct.collect() //最多有500个分区
    partitions.grouped(5).foreach(batches=>{
        batches.par.foreach(row=>{
          val partitionKey = row.getAs[Double]("partition_column")
          val writeDF = df1.as("df1").join(df2.as("df2"),
            col("df1.partition_column").equalTo(partitionKey)
            and col("df2.partition_column").equalTo(partitionKey)
            and col("df1.unique_id").equalTo(col("df2.unique_id"))).select("df1.*")
writeDF.write.mode(SaveMode.Overwrite).parquet(s"somebase_location/partition_column=${partitionKey}/")
        })
    })

该方案执行计划同样使用分区过滤，资源利用率约75%（方案1仅30%）。

疑问 #

1. 为什么方案2比方案1更快？
2. 由于仅需df1的数据，是否应采用过滤方案？

Spark UI分析结果（2023-06-24编辑） #

方案1细节 #

1. 若为Parquet Hive表，Spark默认会读取所有分区；添加目标分区过滤条件后，仅读取指定分区，生成两个作业：一个处理数据，一个写入结果。可开启配置项spark.sql.hive.metastorePartitionPruning，将部分谓词下推至Hive元数据提前排除不匹配分区（暂未测试）。优化后的代码如下：

val partitions = df2.select("partition_column").distinct.collect()
val partitionsKeys = partitions.map(key=>key.getInt(0))
val writeDF = df1.as("df1").join(df2.as("df2"),
col("df1.partition_column").isin(partitionsKeys:_*)  
and  col("df1.partition_column").equalTo("df2.partition_column")
            and col("df1.unique_id").equalTo(col("df2.unique_id"))).select("df1.*")
writeDF.write.mode(SaveMode.Overwrite).partitionBy("partition_column").parquet("some location")

2. 读取数据后会对所有记录排序并执行Shuffle，Shuffle写入量达TB级。
3. 最终执行SortMerge Join后写入数据。

方案2细节 #

1. Spark以5个分区为一组，逐个读取分区执行Join，生成的作业数量多于方案1。
2. 对数据排序后执行Broadcast Join（因单分区下df2数据量极小），最终写入数据。

解答 #

问题1：方案2更快的原因 #

• Join策略更优：方案2针对单个分区Key过滤后，df2的单分区数据量极小，Spark自动选择Broadcast Join，避免了方案1中SortMerge Join带来的大规模Shuffle（TB级写入量），Shuffle开销大幅降低。
• 资源利用率更高：方案1的SortMerge Join需要大量资源进行全局排序和Shuffle，易出现资源分配不均或等待情况；方案2将任务拆分为多个小作业，每个小作业计算量更均衡，能更充分利用集群资源（利用率从30%提升至75%）。
• 分区读取更精准：方案2直接指定单个分区Key读取数据，避免了方案1中未开启元数据分区修剪时的全分区扫描，减少了不必要的数据读取开销。

问题2：是否应采用过滤方案 #

是的，建议采用基于分区过滤的优化方案，理由如下：

• 需求本质是从df1中过滤出与df2匹配的记录，可以用半连接（Semi Join）替代普通Join，进一步减少数据处理量：

val df2Filter = df2.select("partition_column", "unique_id").distinct()
val filteredDF1 = df1.join(df2Filter, 
    Seq("partition_column", "unique_id"), 
    "leftsemi")
filteredDF1.write.mode(SaveMode.Overwrite).partitionBy("partition_column").parquet("some location")

• 半连接仅返回df1中匹配的记录，无需处理df2的额外数据，比普通Join更高效；同时结合分区过滤（提前获取df2的分区Key列表，过滤df1的分区），能进一步减少数据读取量。
• 若开启spark.sql.hive.metastorePartitionPruning=true，Spark会直接从Hive元数据层面过滤掉不需要的分区，避免全分区扫描，性能会更优。

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