Spark Streaming trigger.once()场景下如何获取每个微批处理的记录数

你遇到的问题主要来自3个层面：

• 变量作用域问题：Python函数内部对全局变量赋值时默认会将变量识别为局部变量，你代码中write_to_managed_table函数内的ss_count没有声明为全局变量，赋值操作不会修改函数外定义的全局ss_count，外部变量始终为初始值0。
• 执行时序问题：writeStream.start()是异步执行方法，调用后只会在后台启动流作业，后续的print(ss_count)会立刻执行，此时微批处理逻辑还未运行，自然拿不到统计结果。
• 逻辑缺失问题：当前write_to_managed_table函数内仅做了计数，没有实际写入Delta表的逻辑，作业不会完成你预期的数据落库动作。

方案1：使用Spark累加器实现跨节点计数（适合需要在Driver端后续使用计数结果的场景） #

Spark累加器是专门设计的跨节点安全累加的变量，完全适配分布式场景下的统计需求，示例代码如下：

# 初始化Long类型累加器
ss_count = spark.sparkContext.accumulator(0)

def write_to_managed_table(micro_batch_df, batchId):
    # 先缓存微批数据，避免count和写表两次执行计算
    micro_batch_df.cache()
    # 累加计数
    batch_count = micro_batch_df.count()
    ss_count.add(batch_count)
    # 打印审计日志
    print(f"batch_id:{batchId}, 写入记录数: {batch_count}, 本次作业累计写入记录数: {ss_count.value}")
    # 实际写入Delta表的逻辑
    micro_batch_df.write.format("delta").mode("append").save(saveloc)
    # 释放缓存
    micro_batch_df.unpersist()

saveloc = "TABLE_PATH"
query = df_final.writeStream.trigger(once=True)\
    .foreachBatch(write_to_managed_table)\
    .option('checkpointLocation', f"{saveloc}/_checkpoint")\
    .start()
# 等待流作业执行完成再读取累加器值
query.awaitTermination()
print(f"本次作业总写入记录数: {ss_count.value}")

方案2：直接写入审计日志/审计表（适合需要持久化审计记录的场景） #

如果不需要在Driver后续逻辑中使用计数值，仅需留存审计记录，可以直接在foreachBatch中将计数结果写入单独的审计表，和业务表写入操作放在同一个事务中，保证数据和审计记录的一致性：

def write_to_managed_table(micro_batch_df, batchId):
    micro_batch_df.cache()
    batch_count = micro_batch_df.count()
    # 写业务表
    micro_batch_df.write.format("delta").mode("append").save(saveloc)
    # 写审计表
    audit_df = spark.createDataFrame([(batchId, batch_count, current_timestamp())], 
                                    ["batch_id", "write_count", "write_time"])
    audit_df.write.format("delta").mode("append").save("AUDIT_TABLE_PATH")
    micro_batch_df.unpersist()

方案3：基于Delta表历史信息统计（无需修改现有写入逻辑的无侵入方案） #

Delta本身会记录表的版本变更历史，你可以在每次流作业运行完成后，直接查询Delta表的版本历史获取本次写入的记录数：

from delta.tables import DeltaTable

# 先运行流作业
saveloc = "TABLE_PATH"
query = df_final.writeStream.trigger(once=True)\
    .format("delta")\
    .option('checkpointLocation', f"{saveloc}/_checkpoint")\
    .start(saveloc)
query.awaitTermination()

# 查询Delta表最近一次版本的新增记录数
delta_table = DeltaTable.forPath(spark, saveloc)
last_version = delta_table.history().select("version", "numOutputRows").orderBy("version", ascending=False).first()
print(f"本次写入版本号: {last_version['version']}, 写入记录数: {last_version['numOutputRows']}")

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