Spark Structured Streaming(3.x)如何查看每批次处理的记录数量

Spark Structured Streaming 3.x版本完全支持查看每批次处理记录数指标，对应传统Spark Streaming的同类指标，可通过以下3种常用方式获取：

• Spark UI直接查看
  提交Structured Streaming作业后，进入Spark UI的Structured Streaming标签页，选择你要查看的流式查询，在批次统计列表中即可看到每批次对应的numInputRows（单批次总输入记录数）、inputRowsPerSecond（输入速率）、processedRowsPerSecond（处理速率）字段，其中numInputRows就是你需要的每批次处理记录总数指标。
• 通过内置Metrics系统采集
  先在Spark作业配置中开启流式指标上报：将spark.sql.streaming.metricsEnabled参数设为true。开启后Spark会通过Dropwizard指标体系暴露相关指标，你可以对接Prometheus、Grafana等监控系统采集[查询唯一标识].numInputRows指标，即可持续获取每批次的处理记录数。
• 自定义监听代码获取
  你可以通过实现StreamingQueryListener接口自定义监听逻辑，直接在代码中拿到每批次的记录数，示例代码如下：

  import org.apache.spark.sql.streaming.StreamingQueryListener
  import org.apache.spark.sql.streaming.StreamingQueryListener._
  
  spark.streams.addListener(new StreamingQueryListener() {
    override def onQueryStarted(event: QueryStartedEvent): Unit = {}
    override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {}
    override def onQueryProgress(event: QueryProgressEvent): Unit = {
      // 直接获取当前批次总输入记录数
      val batchTotalRows = event.progress.numInputRows
      // 这里可自行添加日志打印、写入外部存储等逻辑
      println(s"当前批次ID: ${event.progress.batchId}, 处理记录总数: $batchTotalRows")
    }
  })
  

  如果你的作业有多个输入源，还可以从event.progress.sources数组中单独获取每个输入流对应的每批次记录数。

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