Spark查询下推在Join场景未正常工作问题排查求助

我正在编写从Snowflake取数的Spark程序，涉及两张记录数均<50000条表的简单Join查询。单独执行单表查询和直接在Snowflake中执行完整Join都很快，但Spark程序执行耗时极长，已捕获Spark物理计划（截图略）。

原程序代码 #

val TABB = spark.sql("SELECT COL1, COL2 FROM XYZ WHERE DATE(COL2) >= DATE(ADD_MONTHS((DATE_ADD(CURRENT_DATE(), -(DAY(CURRENT_DATE())-1))),-3)) ")

TABB.createOrReplaceTempView("TAB1")

val TABA = spark.sql("SELECT COL1, 'ABC'||COL3 AS COL4 FROM PQR WHERE YEAR(COL3)=2010 ")

TABA.createOrReplaceTempView("TAB2")

val result = spark.sql("SELECT A.COL1, COL2, COL3, COL4 FROM TAB1 A LEFT JOIN TAB2 B ON A.COL1=B.COL1 ")

result.count

问题根源 #

当前写法是先将两张表的数据分别拉取到Spark集群，再在Spark端执行Join操作。即便单表数据量不大，Spark的Shuffle Join会引发跨节点数据传输、分区重组等开销，效率远低于让Snowflake（原生数据存储端）直接完成Join后返回结果。

解决方案 #

1. 将完整查询逻辑下推到Snowflake
   把所有SQL逻辑合并为一条查询，让Snowflake执行Join、过滤等操作，仅返回最终结果，避免Spark端的Shuffle开销：

   val result = spark.sql("""
     SELECT A.COL1, A.COL2, B.COL3, 'ABC'||B.COL3 AS COL4 
     FROM (
       SELECT COL1, COL2 
       FROM XYZ 
       WHERE DATE(COL2) >= DATE(ADD_MONTHS(DATE_ADD(CURRENT_DATE(), -(DAY(CURRENT_DATE())-1)), -3))
     ) A 
     LEFT JOIN (
       SELECT COL1, COL3 
       FROM PQR 
       WHERE YEAR(COL3)=2010
     ) B ON A.COL1=B.COL1
   """)
   result.count
   

2. 启用Spark-Snowflake的查询下推优化
   检查并配置以下Spark参数，确保连接器尽可能将查询逻辑下推到Snowflake执行：

   spark.sql.sources.pushDownAggregate=true
   spark.sql.sources.pushDownPredicate=true
   spark.sql.sources.pushDownJoin=true
   

3. 优化Spark端Join策略（若必须在Spark执行）
   若业务场景要求必须在Spark端执行Join，可通过广播小表避免Shuffle：

   import org.apache.spark.sql.functions.broadcast
   import org.apache.spark.sql.functions.expr
   
   val tab1 = spark.sql("SELECT COL1, COL2 FROM XYZ WHERE DATE(COL2) >= DATE(ADD_MONTHS(DATE_ADD(CURRENT_DATE(), -(DAY(CURRENT_DATE())-1)), -3))")
   val tab2 = spark.sql("SELECT COL1, COL3 FROM PQR WHERE YEAR(COL3)=2010")
   
   val result = tab1.join(broadcast(tab2), Seq("COL1"), "left")
                    .select(tab1("COL1"), "COL2", "COL3", expr("'ABC'||COL3 AS COL4"))
   result.count
   

   可通过result.explain(true)查看物理计划，确认是否使用了Broadcast Hash Join。

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