数据源 API 与谓词下推：让存储层帮忙过滤

上一篇解决了 AQE 的运行时优化。这一篇进入数据源层——Spark 如何把计算推到存储层执行。

查询优化不只发生在执行引擎内部。如果存储层能在读取数据时就过滤掉不需要的行和列，引擎需要处理的数据量会大幅减少。Spark 通过 DataSource API 和 Catalyst 的优化规则把这个能力标准化了：谓词下推（Predicate Pushdown）让存储层只返回满足条件的行，列裁剪（Column Pruning）让存储层只返回查询需要的列。

本文只抓一个问题：谓词下推和列裁剪如何从 Catalyst 的逻辑计划传递到数据源实现。

DataSource API 的两代演进

Spark 的数据源 API 经历了两代：

DataSource V1 (Spark 1.3+):
  接口: InputFormat / OutputFormat + createRelation
  谓词下推: 通过 PrunedFilteredScan trait 可选实现
  问题: API 不够灵活，不支持流式读写，难以做细粒度优化

DataSource V2 (Spark 2.3+, 重构于 3.0):
  接口: TableProvider → Table → ScanBuilder → Scan → Batch → PartitionReader
  谓词下推: SupportsPushDownFilters 接口
  列裁剪: SupportsPushDownRequiredColumns 接口
  优势: 统一批/流读写、支持事务、细粒度优化协商

V2 API 的三层抽象——Table 描述数据的 schema 和能力，ScanBuilder 协商优化（接收 Catalyst 下推的谓词和列列表），Scan 和 PartitionReader 执行实际读取。

关键设计：下推是协商式的，不是命令式的。Catalyst 把谓词列表发给 ScanBuilder，ScanBuilder 返回它能处理的谓词子集。不能处理的谓词仍然由 Spark 在引擎层面执行 Filter。

Catalyst Optimizer
    │ 收集 Filter 节点中的所有谓词
    ▼
ScanBuilder.pushFilters([age > 18, name LIKE '%test%'])
    │ 数据源检查能力
    ▼
返回: 已下推 [age > 18], 未下推 [name LIKE '%test%']
    │
    ▼
物理计划:
  Filter (name LIKE '%test%')    ← Spark 执行
  └── Scan (pushed: age > 18)   ← 数据源执行

Parquet 的行组级过滤

Parquet 是 Spark 最常用的列式存储格式。Parquet 文件内部按行组（Row Group，默认 128MB）组织数据，每个行组包含列块（Column Chunk），每个列块有统计信息（min/max/null_count）。

当 Spark 把 age > 18 下推到 Parquet 时，Parquet Reader 在读取每个行组前先检查该行组 age 列的 min/max 统计：

Row Group 0: age min=5,  max=15  → age > 18 不可能满足 → 跳过整个行组
Row Group 1: age min=10, max=25  → 可能满足 → 读取并逐行过滤
Row Group 2: age min=20, max=30  → age > 18 全部满足 → 全部读取
Row Group 3: age min=1,  max=8   → 跳过

一个 100 行组的 Parquet 文件，如果谓词能跳过 80% 的行组，Spark 只需要读取 20% 的数据。对于大规模数据集，这个优化可以把 I/O 量降低一个数量级。

列裁剪同样发挥作用：如果查询只需要 name 和 age 两列，Parquet Reader 只读取这两列的列块，跳过其他列。Parquet 的列式存储让这个优化几乎零代价——不同列的数据在文件中是物理分离的。

ORC 和 Delta Lake 的类似机制

ORC 文件有类似的 Stripe 级统计和 Bloom Filter 索引，谓词下推的工作方式和 Parquet 基本相同。

Delta Lake 在 Parquet 之上增加了 Z-Order 索引和数据跳过（Data Skipping）机制。OPTIMIZE ZORDER BY (city, date) 把数据按多维空间排列，使得 WHERE city = ‘Beijing’ AND date > ‘2024-01-01’ 可以跳过大量无关的 Parquet 文件。

实验：观察谓词下推

val spark = SparkSession.builder.getOrCreate()

// 准备 Parquet 测试数据
val data = spark.range(10000000)
  .withColumn("name", concat(lit("user_"), ($"id" % 1000).cast("string")))
  .withColumn("age", ($"id" % 100).cast("int"))
data.write.mode("overwrite").parquet("/tmp/users.parquet")

// 查询并检查下推
val df = spark.read.parquet("/tmp/users.parquet")
val result = df.filter($"age" > 80).select($"name", $"age")

// 查看物理计划
result.explain(true)
// 在 Physical Plan 中找 FileScan parquet 节点
// PushedFilters: [IsNotNull(age), GreaterThan(age,80)]
// ReadSchema: struct<name:string,age:int>  ← 只读 2 列，不读 id 列

// 对比无下推的全表扫描
val fullScan = df.select($"name", $"age", $"id")
fullScan.explain(true)
// ReadSchema 包含所有 3 列

在 explain 输出中，PushedFilters 列表显示了成功下推到数据源的谓词。ReadSchema 显示了列裁剪后实际读取的列。如果某个谓词没有出现在 PushedFilters 中，说明数据源不支持该类型的谓词下推（比如某些 UDF 谓词）。

在 Spark UI 的 Stage 详情页中，查看 Scan 算子的 “number of files read” 和 “number of output rows” 指标，可以量化跳过了多少数据。

模式提炼

谓词下推体现了计算下沉模式：把过滤计算尽量推到数据存储层执行，减少传输和处理的数据量。

这个模式的应用范围远超 Spark。数据库的索引查询就是最经典的谓词下推——WHERE 条件通过 B+ 树索引定位数据页，而不是全表扫描后过滤。分布式数据库（如 TiDB）把谓词下推到存储节点（TiKV），让存储节点在返回数据前完成过滤。CDN 的边缘计算把部分处理逻辑下推到边缘节点，减少回源流量。

通用原则：数据移动比计算移动贵。如果能在数据所在的地方完成过滤，就不要先把数据搬到计算节点再过滤。

工程迁移表

Spark 概念	传统数据库对应	Flink 对应	分布式存储对应
谓词下推	索引查询 / Index Scan	FilterableTableSource	存储层的 Bloom Filter
列裁剪	投影操作（只读需要的列）	ProjectableTableSource	列式存储的选择性读取
DataSource V2 API	Storage Engine 接口	DynamicTableSource	存储插件接口
行组级统计（Parquet）	页级统计 / Zone Map	无直接对应	LSM-Tree 的 Block 级过滤
分区裁剪	分区表的分区消除	Partition Discovery	目录级文件跳过

常见误解

误解一：“所有 Filter 都能下推”。只有数据源声明支持的谓词类型才能下推。复杂的 UDF 谓词、涉及多列的计算谓词通常不能下推。Spark 的 explain 输出中，PostScan Filter 显示了未下推的谓词。

误解二：“谓词下推自动保证减少 I/O”。如果数据没有排序或索引，下推的谓词可能无法跳过任何行组。比如 age 列的值在每个行组中都分布在 0-100 之间，age > 80 无法跳过任何行组。数据的物理布局（排序、分桶、Z-Order）决定了下推的实际效果。

误解三：“列裁剪对行式存储也有效”。列裁剪对列式存储（Parquet/ORC）效果显著，因为不同列的数据在文件中物理分离。对于行式存储（CSV/JSON），即使 Spark 只需要两列，也必须读取完整的行然后丢弃多余的列。

练习

用 spark.read.parquet 读取一个多列 Parquet 文件，分别执行 select("col1") 和 select("*")，在 Spark UI 中对比 Scan 的数据读取量（Input Size / Records）。
创建一个 Parquet 文件，数据按 age 列排序后写入。然后对比 WHERE age > 80 在排序数据和未排序数据上的行组跳过数量。用 spark.conf.set("parquet.filter.statistics.enabled", "true") 确保统计过滤开启。
写一个自定义 UDF 作为 Filter 条件（如 udf_check(age)），用 explain 观察这个谓词是否出现在 PushedFilters 中。思考为什么 UDF 无法下推。

系列导航

序号	主题	状态
00	导读：为什么 Spark 的核心是一张 DAG	已完成
01	RDD：弹性分布式数据集的五大属性	已完成
02	宽依赖与窄依赖：Stage 是怎样划分出来的	已完成
03	DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行	已完成
04	Shuffle 机制：数据跨分区交换的代价与优化	已完成
05	内存管理与 Tungsten：堆外内存、序列化与代码生成	已完成
06	存储体系：Block Manager、广播变量与累加器	已完成
07	容错机制：Lineage、Checkpoint 与推测执行	已完成
08	Spark SQL 与 Catalyst：从 SQL 文本到物理算子树	已完成
09	DataFrame 与 Dataset：类型安全与性能的折中	已完成
10	Adaptive Query Execution：运行时改写执行计划	已完成
11	数据源 API 与谓词下推：让存储层帮忙过滤	本篇
12	Structured Streaming：把流当成无界表	下一篇
13	水位线与状态管理：流处理的时间语义
14	Spark on YARN / K8s：资源管理与部署模式
15	动态资源分配与作业调度：多租户集群的资源博弈
16	性能诊断：Spark UI、Event Log 与调优思路
17	Spark 与 Flink：两种计算引擎的设计选择