运行中PySpark流作业如何为Delta表新增列并写入逐批递增batch_Id

• 方案1（直接通过withColumn加lit(generate_id())）失效原因：Structured Streaming的查询逻辑在作业启动时就会完成编译，lit(generate_id())属于驱动侧启动时就计算完成的常量值，后续所有批次运行时都直接复用这个固定值，不会每个批次重新执行generate_id函数，因此持续运行时所有批次ID相同。
• 方案2（foreachBatch写入报数据源不匹配）错误原因：foreachBatch本身就是流式作业的输出接收器，其传入的处理函数中拿到的是普通批式DataFrame，若在该函数内继续使用流式写入API（如writeStream、流式toTable方法）写入Delta表，就会触发接收器类型与目标表数据源不匹配的AnalysisException。

核心逻辑：将batch_Id生成、自增的逻辑完全放入foreachBatch的微批处理函数中，每个批次触发时才执行ID计算，微批内使用批式API写入Delta表即可。
参考实现代码：

from pyspark.sql.functions import lit

# 初始化批次起始ID，作业重启时可先查询目标表已存在的最大Batch_Id+1作为起始值，避免ID重复
current_batch_id = 0

def batch_process(batch_df, _):
    global current_batch_id
    # 为当前批次所有行添加统一Batch_Id
    processed_df = batch_df.withColumn("Batch_Id", lit(current_batch_id))
    # 用批写入API写入Delta表，禁止在foreachBatch内部使用流式写入方法
    processed_df.write \
        .format("delta") \
        .mode("append") \
        .option("mergeSchema", "true") \
        .saveAsTable("patients")
    # 写入成功后再自增ID，避免写入失败导致ID跳号
    current_batch_id += 1

# 配置CSV流读取
csv_stream = spark.readStream \
    .format("csv") \
    .option("header", "true") \
    .option("maxFilesPerTrigger", 1) \
    .load("abfss://<your-container>@<your-adls-account>.dfs.core.windows.net/<csv-storage-path>")

# 启动流作业
stream_query = csv_stream.writeStream \
    .foreachBatch(batch_process) \
    .option("checkpointLocation", "abfss://<your-container>@<your-adls-account>.dfs.core.windows.net/<checkpoint-path>") \
    .start()

stream_query.awaitTermination()

注意事项 #

• 必须配置独立的checkpoint路径，否则作业重启后会重复读取已处理的CSV文件
• 若生产环境需要保证作业重启后batch_Id连续不重复，初始化current_batch_id时先执行spark.sql("select max(Batch_Id) from patients").collect()[0][0]获取历史最大ID，在此基础上+1作为起始值即可
• 若CSV表结构存在变动，可在写入时开启mergeSchema选项自动同步Schema到Delta表

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