Spark不同物理计划评估及优化相关问题咨询

测试背景 #

近期为调优现有Spark流程，基于Delta表开展了查询计划的学习与测试：

原自连接方案 #

通过自连接获取最新数据，代码如下：

table1 = spark.read.format('delta').load('somepath')
df1 = table1.where("columna1 = 'x' and column2 = 'y'")
df2 = table1.groupBy("column4","column5").agg(max("DATE").alias("Date"))

df1 = df1.join(df2, (df1.con1 == df2.cond1) & (df1.con2 == df2.con2) & (df1.con3 == df2.con3) , how = "inner").select("col1", "col2", "col3","col4")

物理计划包含SortMergeJoin、多次Exchange操作，处理数据量达3.6 TiB。

Window函数优化方案 #

改用Window函数重构逻辑，代码如下：

table1 = spark.read.format('delta').load('somepath').where("columna1 = 'x' and column2 = 'y'")
windowSpec = Window.partitionBy("rule1", "rule2").orderBy(col("rule1").desc())
final = table1.withColumn("recent_value", max("DATE").over(windowSpec)) \
                        .filter(col("DATE") == col("recent_value")).select("col1", "col2", "col3","col4")

物理计划更简洁，处理数据量仅20.1 MiB。后续还尝试用Coalesce替换Exchange、调整select执行顺序，得到不同物理计划。

问题解答 #

1. Window方案是否优于原自连接方案？ #

是的，从测试数据就能直接验证：Window方案处理的数据量从3.6 TiB骤降到20.1 MiB，物理计划也更简洁。
原因在于自连接方案需要先对全表做分组聚合生成df2，再和过滤后的df1执行SortMergeJoin，这个过程会触发多次数据shuffle（对应Exchange操作），带来大量数据传输与磁盘IO开销；而Window函数可以在同一个分区内完成最大值计算和筛选，避免了跨分区的shuffle，尤其是当分组键（partitionBy的字段）基数不高时，性能提升会非常显著。

2. Window场景下，用Coalesce替代Exchange是否更优？ #

没有绝对的答案，得结合具体场景判断：

• 如果目标是减少输出文件数量，且Window处理后的数据分布相对均匀，用Coalesce更优——因为Coalesce是在现有分区基础上直接合并，不会触发shuffle，开销极小；
• 但如果Window操作后的数据分布极不均匀（比如部分分区数据量极大），Exchange（即repartition）能重新均匀分配数据，避免后续操作出现数据倾斜，这种情况下Exchange更合适。
  另外要注意，Window操作本身依赖分区布局，如果强行用Coalesce合并过多分区，可能导致单个分区数据量过大，反而拖慢Window内的计算逻辑。

3. 当筛选依赖的列属于最终需要的列时，先select还是先filter更优？ #

优先先执行filter再做select。
哪怕筛选依赖的列是最终需要保留的，先做过滤能提前缩减数据行数——虽然Spark优化器会自动做谓词下推，但手动提前执行filter能确保过滤逻辑尽早生效，减少后续操作（包括select）需要处理的数据量，从而降低内存占用和IO开销。比如先筛选出符合条件的行，再选择需要的列，比先选列再过滤要少处理大量不必要的行数据。

4. 查询计划中Project操作（对应select）放在开头还是结尾更优？有哪些判断技巧？ #

没有绝对最优的位置，核心看数据量缩减的时机和操作的依赖关系：

• 如果Project操作能大幅减少数据宽度（比如只选少数几列，原表列数多且包含大字段），且这些列能覆盖后续所有操作的依赖（过滤、聚合、Window等都只用到这些列），那放在开头更优——提前减少每一行的数据大小，降低内存占用和数据传输量；
• 如果后续操作（比如filter、Window）需要用到原表中不在最终Project列里的字段，那必须把Project放在结尾，否则会因缺少依赖字段导致逻辑错误。

判断技巧：

• 先梳理所有操作的依赖字段：如果后续所有操作只需要最终select的列，就把Project提前；
• 评估原表的数据体积：如果原表有大量大字段（比如长字符串、二进制数据），提前Project能显著减少数据体积；
• 结合Spark优化器逻辑：虽然Catalyst会自动做列裁剪，但手动提前Project能更明确地引导优化器，避免特殊场景下的优化失效。

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