Spark Structured Streaming Query能否统计启动以来累计处理总行数？

Great question! Let's break this down clearly:

核心结论 #

是的，Spark Structured Streaming没有直接提供开箱即用的“自启动以来累计处理总行数”指标——你确实需要通过StreamingQueryListener自行跟踪计算，但实现起来非常简单，而且有替代方案可选。

为什么需要自行跟踪？ #

Spark内置的Structured Streaming指标（比如通过query.lastProgress获取的信息）主要聚焦在单个批次的处理细节（如当前批次处理行数、延迟、速率等）。而累计总行数属于跨批次的聚合统计，这部分状态Spark不会自动帮你维护和暴露，所以需要我们手动实现。

实现方式：自定义StreamingQueryListener #

你可以通过自定义StreamingQueryListener，在每个批次完成后捕获该批次的处理行数，然后累加得到累计总数。这里要注意线程安全，因为Listener的回调会在不同线程执行。

Scala示例 #

import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener.{QueryProgressEvent, QueryStartedEvent, QueryTerminatedEvent}
import java.util.concurrent.atomic.AtomicLong

// 线程安全的累计计数器
val totalRowsProcessed = new AtomicLong(0L)

// 自定义Listener
val queryListener = new StreamingQueryListener() {
  override def onQueryStarted(event: QueryStartedEvent): Unit = {}

  override def onQueryProgress(event: QueryProgressEvent): Unit = {
    // 获取当前批次处理的行数
    val batchRows = event.progress.numInputRows
    // 原子性累加，避免并发问题
    totalRowsProcessed.addAndGet(batchRows)
    // 可根据需求打印或推送指标到监控系统
    println(s"📊 累计处理总行数: ${totalRowsProcessed.get()}")
  }

  override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {}
}

// 注册Listener到SparkSession
spark.streams.addListener(queryListener)

Python示例 #

from pyspark.sql.streaming import StreamingQueryListener
import threading

# 线程安全的计数器与锁
total_rows_processed = 0
counter_lock = threading.Lock()

class CustomStreamingQueryListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        pass

    def onQueryProgress(self, event):
        global total_rows_processed
        batch_rows = event.progress.numInputRows
        # 加锁保证并发安全
        with counter_lock:
            total_rows_processed += batch_rows
        print(f"📊 累计处理总行数: {total_rows_processed}")

    def onQueryTerminated(self, event):
        pass

# 注册Listener
spark.streams.addListener(CustomStreamingQueryListener())

进阶：持久化累计值（可选） #

如果你的Streaming Query需要在重启后恢复累计数，那么需要把累计值持久化到外部存储（比如Redis、关系型数据库）：

• 每次批次完成后，把更新后的累计数写入外部存储
• Query启动时，先从外部存储读取最新的累计值作为初始值

替代方案：借助监控系统聚合 #

如果你已经有监控栈（比如Prometheus + Grafana），可以不用在代码里维护累计值：

1. Spark的Metrics系统会自动暴露每个批次的numInputRows指标
2. 在监控系统中对该指标做累计求和（比如Prometheus的sum_over_time函数），就能得到自Query启动以来的总行数

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