如何在Spark中禁用扫描后的谓词下推优化

我尝试禁用Spark的一项优化。从物理执行计划和日志可见，Spark在ORC扫描（Scan orc）后添加了包含array_intersect的Post-Scan Filter，该操作在后续Project步骤中重复执行，导致作业耗时增加50%。使用.cache()可避免该下推并提速，但数据集过大（磁盘占用达100TB）无法采用。尝试配置spark.sql.optimizer.excludedRules排除PushDownPredicate无效，请问如何不使用cache禁用该扫描后过滤？

物理执行计划 #

== Physical Plan ==
AdaptiveSparkPlan (11)
+- Project (10)
   +- Project (9)
      +- BroadcastHashJoin Inner BuildRight (8)
         :- Project (3)
         :  +- Filter (2)
         :     +- Scan orc  (1)
         +- BroadcastExchange (7)
            +- Project (6)
               +- Filter (5)
                  +- Scan orc  (4)


(1) Scan orc 
Output [2]: [id#70, segment#71]
Batched: false
Location: InMemoryFileIndex [file:/....]
PushedFilters: [IsNotNull(id)]
ReadSchema: struct<id:string,segment:array<struct<ids:bigint>>>

(2) Filter
Input [2]: [id#70, segment#71]
Condition : ((size(array_intersect(segment#71.ids, [100,200,300,400]), true) > 0) AND isnotnull(id#70))

日志信息 #

23/08/03 09:06:50 INFO FileSourceStrategy: Pushed Filters: IsNotNull(id)
23/08/03 09:06:50 INFO FileSourceStrategy: Post-Scan Filters: (size(array_intersect(segment#71.ids, [100,200,300,400]), true) > 0),isnotnull(id#70)

作业代码 #

spark.read.orc(s"input")
  .select(
    col("id"),
    col("segment.ids").as("segment_ids"),
  )
 .join(otherSmallDf, Seq("id"), "inner")
 .select(
    someUDF(col("id")).as("userId"),
    col("segment_ids")
  )
 .withColumn("segment_ids", array_intersect(col("segment_ids"), shareableSegmentIds))
 .where(size(col("segment_ids")) > 0)
 .write.option("sep", "\t").csv("output/path")

解决方案 #

1. 用临时视图添加优化屏障，阻止过滤下推 #

通过创建临时视图，让Spark将数据读取和后续处理视为两个独立阶段，打断优化器的下推逻辑：

// 读取数据并创建临时视图
spark.read.orc(s"input")
  .select(
    col("id"),
    col("segment.ids").as("segment_ids"),
  )
 .createOrReplaceTempView("temp_input")

// 从临时视图读取后执行后续逻辑
spark.sql("SELECT id, segment_ids FROM temp_input")
 .join(otherSmallDf, Seq("id"), "inner")
 .select(
    someUDF(col("id")).as("userId"),
    col("segment_ids")
  )
 .withColumn("segment_ids", array_intersect(col("segment_ids"), shareableSegmentIds))
 .where(size(col("segment_ids")) > 0)
 .write.option("sep", "\t").csv("output/path")

2. 排除更精准的优化规则 #

PushDownPredicate是顶层规则，实际生成Post-Scan Filter的是PushDownFilters和数据源级别的PushDownPredicate，可尝试同时排除这两个规则（注意Spark版本差异，以下为Spark 3.x路径）：

spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.PushDownFilters,org.apache.spark.sql.execution.datasources.PushDownPredicate

3. 拆分读取与join，延迟嵌套字段解析 #

通过拆分数据读取步骤，先只读取id字段完成join，再关联原表获取嵌套字段，让过滤逻辑仅在join后执行：

import org.apache.spark.sql.types._

val customSchema = StructType(Seq(
  StructField("id", StringType),
  StructField("segment", StructType(Seq(StructField("ids", ArrayType(LongType)))))
))

// 先读取id字段完成join
val joinedIds = spark.read.schema(customSchema).orc(s"input")
  .select(col("id"))
 .join(otherSmallDf, Seq("id"), "inner")

// 关联原表获取segment字段，再执行过滤逻辑
joinedIds.join(
    spark.read.schema(customSchema).orc(s"input").select(col("id"), col("segment.ids").as("segment_ids")),
    Seq("id")
  )
 .select(
    someUDF(col("id")).as("userId"),
    col("segment_ids")
  )
 .withColumn("segment_ids", array_intersect(col("segment_ids"), shareableSegmentIds))
 .where(size(col("segment_ids")) > 0)
 .write.option("sep", "\t").csv("output/path")

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