Spark Streaming指定分区读取异常：无限运行无数据的解决方法

问题场景与现象 #

在Databricks上搭建PySpark Streaming管道，源Delta表按RECORD_SOURCE、ORDER_YEAR、ORDER_MONTH分区，使用availableNow触发器，通过Spark Streaming Checkpoint跟踪新数据，需求是仅读取SOURCE_2分区（该分区每日更新3次，SOURCE_1分区为实时更新）。

当前问题：

• 管道无限运行但未加载任何数据，批次统计显示numInputRows=0，但numFilesOutstanding数值非零
• 即使SOURCE_2无更新，重启流任务后numFilesOutstanding仍持续增长，推测流在尝试读取SOURCE_1分区的文件

相关代码：

(
    spark.readStream.format("delta")
    .option("ignoreChanges", "true")
    .option("maxBytesPerTrigger", self.bytes_to_process)
    .option("maxFilesPerTrigger", self.files_to_process)
    .table(self.silver_table)
    .filter(col("RECORD_SOURCE") == "SOURCE_2")
    .filter(f"CONCAT(ORDER_YEAR, ORDER_MONTH) >= '202306'")
    .writeStream.format("delta")
    .foreachBatch(self.process_streaming_batch)
    .option("checkpointLocation", self.checkpoint)
    .trigger(availableNow=True).start()
)

已尝试的操作：

• 删除Checkpoint后重启任务，首次运行正常，但后续问题复现
• 将ignoreChanges替换为DBR 12.2推荐的skipChangeCommits，问题依旧

问题解答 #

1. 该场景是否常见，此行为是否符合预期？ #

这种场景并不罕见，但行为不符合预期。
原因在于：PySpark Streaming读取Delta表时，默认会扫描所有分区的元数据（无论后续是否添加过滤逻辑）。当SOURCE_1这类其他分区持续生成新文件时，流会将这些文件标记为“待处理”（即numFilesOutstanding统计的内容）；即使后续通过filter排除了SOURCE_1的数据，这些文件的元数据仍会被流跟踪，导致numFilesOutstanding持续增长。同时因为过滤后没有符合条件的行，numInputRows为0，流会一直处于等待处理这些“待处理”文件的状态，进而出现无限运行的情况。

2. 有无办法限制流仅读取指定分区？ #

有多种方法可以实现，无需拆分源表：

方法1：读取阶段直接指定分区过滤（推荐）

放弃使用table()方法，改用load()并通过partitionFilters选项指定分区条件，让Spark直接只扫描目标分区的元数据，彻底避免其他分区的干扰。示例代码：

(
    spark.readStream.format("delta")
    .option("skipChangeCommits", "true")
    .option("maxBytesPerTrigger", self.bytes_to_process)
    .option("maxFilesPerTrigger", self.files_to_process)
    # 直接在读取阶段指定分区过滤条件，SQL语法格式
    .option("partitionFilters", "RECORD_SOURCE = 'SOURCE_2' AND CONCAT(ORDER_YEAR, ORDER_MONTH) >= '202306'")
    .load(f"/dbfs/{self.silver_table_path}")  # 替换为你的Delta表实际存储路径
    .writeStream.format("delta")
    .foreachBatch(self.process_streaming_batch)
    .option("checkpointLocation", self.checkpoint)
    .trigger(availableNow=True).start()
)

注意：partitionFilters仅支持针对分区列的SQL条件，无法过滤非分区列。

方法2：确保分区过滤下推生效

确认Spark的分区过滤下推功能开启（默认已开启，可手动配置），让后续的filter逻辑被下推到Delta读取层，减少不必要的分区扫描。可添加以下配置：

spark.conf.set("spark.sql.parquet.filterPushdown", "true")
spark.conf.set("spark.sql.delta.filterPushdown", "true")

这种方式依赖Spark的优化逻辑，可靠性略低于直接指定partitionFilters，尤其是当过滤条件为组合逻辑时。

方法3：辅助优化（可选）

如果存在不需要的非分区列，可以通过includeColumns选项指定仅读取需要的列，减少数据传输开销，但这并非解决分区扫描问题的核心方案：

.option("includeColumns", "需要的列名1, 需要的列名2, RECORD_SOURCE, ORDER_YEAR, ORDER_MONTH")

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