4. 使用PySpark构建批处理管道

在 Databricks 上构建批处理管道，PySpark 的 DataFrame API 是最核心的编程入口。它把结构化数据抽象成一张带 Schema 的分布式表，让开发者用类似 SQL 的声明式写法表达转换逻辑，而由 Catalyst 优化器和 Tungsten 执行引擎负责生成物理计划。理解 DataFrame API 的惰性求值特性，是写好批处理管道的第一步：select、filter、withColumn、groupBy、join 等算子只是构建逻辑计划，真正触发计算的是 write、count、show 这类行动算子。因此，把过滤和列裁剪尽量前移，能显著减少后续 shuffle 与扫描的数据量。

下面用一个订单批处理场景贯穿本节：源表 bronze_orders 保存原始订单，字段包括 order_id、customer_id、order_date、region、amount、status。目标是清洗后按 order_date 分区写入银层表 silver_orders，并支持每日增量追加与全量覆盖两种模式。

from pyspark.sql import functions as F

orders = spark.table("bronze_orders")

cleaned = (orders
    .filter(F.col("status") != "CANCELLED")
    .filter(F.col("amount") > 0)
    .withColumn("order_date", F.to_date("order_date"))
    .withColumn("amount", F.col("amount").cast("decimal(18,2)"))
    .select("order_id", "customer_id", "order_date", "region", "amount", "status")
)

cleaned.write \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .format("delta") \
    .saveAsTable("silver_orders")

这段代码体现了 DataFrame API 的典型写法：先过滤、再转换、最后投影出需要的列。注意 filter 放在最前面，Catalyst 会把它下推到数据源扫描阶段，减少读入行数；select 只保留必要列，也能减少列式存储的读取量。对于 Delta 表，这些优化会直接反映在查询计划的 PushedFilters 和 ReadSchema 中。

分区写入：让数据布局服务于查询

partitionBy("order_date") 会把输出目录按日期切成子目录，例如 silver_orders/order_date=2024-06-01/。分区的价值在于分区裁剪：当查询带有 WHERE order_date = '2024-06-01' 时，引擎只需读取对应目录，而不是全表扫描。分区列的选择要遵循“低基数、查询常用、分布均匀”三个原则。日期、区域、状态适合做分区；order_id 这种高基数列做分区会产生大量小文件，反而拖慢元数据操作和文件列举。

分区数量也需要控制。如果按 order_date 和 region 两级分区，而 region 有 30 个值，一天就会产生 30 个目录。若每天数据量只有几 MB，就会形成大量小文件。此时可以考虑只按日期分区，把 region 留在文件内部作为普通列，或者使用 Delta 的 OPTIMIZE 与 Z-ORDER BY region 来改善局部性。

覆盖与追加：两种写入语义的取舍

mode("overwrite") 会替换目标表的全部数据，适合每日全量重算、维度表刷新或修复历史数据的场景。mode("append") 则只新增数据，适合增量事实表。两者在 Delta 表上的行为差异很关键：overwrite 默认重写整张表，但如果配合 replaceWhere，就可以只覆盖满足条件的分区，实现“分区级覆盖”。

# 只覆盖 2024-06-01 这一天的分区
cleaned.filter(F.col("order_date") == "2024-06-01") \
    .write \
    .mode("overwrite") \
    .option("replaceWhere", "order_date = '2024-06-01'") \
    .partitionBy("order_date") \
    .format("delta") \
    .saveAsTable("silver_orders")

# 增量追加
cleaned.filter(F.col("order_date") == "2024-06-02") \
    .write \
    .mode("append") \
    .partitionBy("order_date") \
    .format("delta") \
    .saveAsTable("silver_orders")

replaceWhere 是批处理管道中非常实用的能力：它让“重跑某一天”变得安全且高效，不会误删其他日期的数据。使用时要确保条件表达式与分区列一致，否则 Delta 会退化为全表覆盖。追加模式则要注意幂等性——如果同一批数据被重复写入，会产生重复行。常见做法是在写入前按业务主键去重，或使用 Delta 的 MERGE INTO 做 upsert。

动态分区裁剪：让过滤条件穿透到扫描层

动态分区裁剪（Dynamic Partition Pruning，DPP）是 Spark 3.0 引入的重要优化。它针对的是分区表 join 场景：当大事实表按分区列 join 小维度表时，Spark 会先从小维度表收集分区键值，再把这些值作为过滤条件下推到事实表扫描，从而只读取匹配的分区。例如事实表 silver_orders 按 order_date 分区，维度表 dim_calendar 只保留最近 7 天，join 后 Spark 会自动只扫描这 7 天的分区。

DPP 生效需要满足几个条件：事实表必须是分区表；join 条件中必须包含分区列；维度表要足够小，能被广播；并且 spark.sql.optimizer.dynamicPartitionPruning.enabled 为 true（默认开启）。在 Databricks 上可以通过 EXPLAIN 查看计划中是否出现 PartitionFilters 或 DynamicPruningExpression 来确认。如果发现没有生效，常见原因是分区列被函数包裹，例如 to_date(order_date) = c.date，这会破坏裁剪条件，应改为直接比较分区列。

把上述能力组合起来，一条可复用的批处理管道通常包含四个环节：读取源表、清洗与规范化、按分区写入目标表、验证分区与行数。验证时可以直接查询 Delta 表的元数据，例如 DESCRIBE DETAIL silver_orders 查看分区数与文件数，或用 SELECT order_date, count(*) FROM silver_orders GROUP BY order_date 确认每个分区的行数是否符合预期。如果发现某个日期分区为空，优先检查源数据中该日期的 status 是否全为 CANCELLED，以及 amount 是否被过滤掉。

最后需要强调的是，DataFrame API 的写法虽然灵活，但批处理管道的稳定性更多取决于数据布局与写入语义的选择。分区列选得对，查询就能裁剪；写入模式选得对，重跑就不会污染历史数据；DPP 生效，join 就不会全表扫描。把这三件事在代码中显式表达出来，比事后调优更有效。