DLT流处理校验列值差异时遇流查询报错的解决方法

方案一：使用DLT中间表+@dlt.expect_all_or_fail（推荐） #

你可以通过声明式中间表完成校验，再从中间表生成丢弃指定列的最终输出表，完美适配流处理且符合DLT的最佳实践：

import dlt
from pyspark.sql.functions import col

# 假设你已经预先定义好schema
# schema = ...

@dlt.table(
    comment="带A、B列相等校验的中间流表"
)
@dlt.expect_all_or_fail(
    {"A_B_must_equal": "A = B"}
)
def validated_raw_stream():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .schema(schema)
        .option("header", "true")
        .option("sep", "|")
        .load(file_path)
    )

@dlt.table(
    comment="最终输出表，已移除A、B列"
)
def final_output_table():
    # 从校验后的中间表读取，直接丢弃A、B列
    return dlt.read("validated_raw_stream").drop("A", "B")

原理说明： #

• @dlt.expect_all_or_fail会自动拦截不符合校验规则的批次（对应单个/多个文件），直接拒绝处理并抛出异常，完全满足你"存在不同值则拒绝文件"的需求。
• 中间表仅作为校验载体保留A、B列，最终表通过drop()移除这两列，既完成校验又满足输出要求，全程兼容流处理逻辑。

方案二：使用foreachBatch手动批次校验（自定义场景） #

如果需要更灵活的错误处理逻辑（比如记录错误详情、自定义报错信息），可以用foreachBatch在每个批次中手动校验：

from pyspark.sql.functions import col
import dlt

def process_batch(df, batch_id):
    # 统计当前批次中A、B列不等的记录数
    invalid_record_count = df.filter(col("A") != col("B")).count()
    if invalid_record_count > 0:
        raise ValueError(f"批次ID {batch_id} 检测到 {invalid_record_count} 条A、B列值不匹配的记录，已拒绝该文件")
    
    # 丢弃A、B列后写入最终表
    df.drop("A", "B").write.mode("append").saveAsTable("your_target_final_table")

# 读取流数据
sourced_stream = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .schema(schema)
    .option("header", "true")
    .option("sep", "|")
    .load(file_path)
)

# 启动流处理，绑定批次处理逻辑
sourced_stream.writeStream.foreachBatch(process_batch).start()

为什么你的原代码报错？ #

流DataFrame是懒加载的连续处理逻辑，不能直接执行count()这类action操作（这类操作会触发同步计算，不符合流处理的异步批次模型），必须通过writeStream系列API触发流的启动和批次处理，所以会出现Queries with streaming sources must be executed with writeStream.start()的报错。

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