如何禁用Spark将Filter下推至SQL Server的JDBC查询优化？

针对你遇到的Spark自动将km_traveled > 1000过滤下推到SQL Server（因该字段无索引导致超时）的问题，以下是几种可行的解决方案：

1. 全局禁用JDBC谓词下推 #

通过Spark配置参数直接关闭所有JDBC数据源的谓词下推优化，这是最直接的方式：

• 在SparkSession初始化时配置：

val spark = SparkSession.builder()
  .appName("CarInfoJob")
  .config("spark.sql.pushdownPredicate.jdbc", "false")
  .getOrCreate()

• 或者在提交作业时通过命令行参数设置：

spark-submit --conf spark.sql.pushdownPredicate.jdbc=false ...

注意：这个配置会禁用所有JDBC读取的谓词下推，可能影响其他需要下推优化的查询性能，适合仅当前作业有此需求的场景。

2. 强制Spark先加载数据再过滤（缓存/临时视图） #

先将符合insertion_time条件的数据全部拉取到Spark集群，再执行过滤操作，避免下推：

方式一：使用缓存 #

// 仅用insertion_time过滤读取数据
val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:sqlserver://your-host:1433;databaseName=your-db")
  .option("dbtable", "(SELECT * FROM car_info WHERE insertion_time > '2022-01-01 00:00:01') AS filtered_car")
  .option("user", "your-user")
  .option("password", "your-pass")
  .load()

// 缓存数据到Spark集群内存/磁盘
df.cache()

// 执行Spark端过滤
val resultDF = df.filter("km_traveled > 1000")

方式二：创建临时视图后查询 #

// 读取数据
val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:sqlserver://your-host:1433;databaseName=your-db")
  .option("dbtable", "(SELECT * FROM car_info WHERE insertion_time > '2022-01-01 00:00:01') AS filtered_car")
  .option("user", "your-user")
  .option("password", "your-pass")
  .load()

// 创建临时视图
df.createOrReplaceTempView("temp_car_info")

// 通过SQL查询执行过滤（此时数据已在Spark端）
val resultDF = spark.sql("SELECT * FROM temp_car_info WHERE km_traveled > 1000")

注意：缓存或临时视图会占用Spark集群的存储资源，需确保集群有足够的内存/磁盘空间容纳insertion_time过滤后的数据集。

3. 使用UDF避免过滤下推 #

将过滤逻辑封装为用户自定义函数（UDF），Spark无法将UDF的逻辑下推到JDBC数据源，因此会在Spark端执行过滤：

import org.apache.spark.sql.functions.udf

// 定义UDF
val filterKmTraveled = udf((km: Int) => km > 1000)

// 读取数据
val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:sqlserver://your-host:1433;databaseName=your-db")
  .option("dbtable", "(SELECT * FROM car_info WHERE insertion_time > '2022-01-01 00:00:01') AS filtered_car")
  .option("user", "your-user")
  .option("password", "your-pass")
  .load()

// 使用UDF过滤
val resultDF = df.filter(filterKmTraveled(df("km_traveled")))

注意：UDF的性能略低于Spark内置的过滤函数，但能精准避免下推特定条件。

4. 针对特定规则禁用Catalyst优化 #

如果你想更精细地控制Catalyst优化规则，可以禁用PushDownPredicate规则，但这会影响所有数据源的谓词下推，而非仅JDBC：

import org.apache.spark.sql.catalyst.optimizer.PushDownPredicate

val spark = SparkSession.builder()
  .appName("CarInfoJob")
  .config("spark.sql.optimizer.excludedRules", "org.apache.spark.sql.catalyst.optimizer.PushDownPredicate")
  .getOrCreate()

注意：这个配置会全局禁用谓词下推优化，影响范围比JDBC专属配置更大，需谨慎使用。

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