如何利用Spark StreamingQueryProgress精准监控结构化流作业性能？

我刚好有过不少用StreamingQueryListener做Spark结构化流监控的实战经验，结合你提到的把指标输出到Kafka的需求，给你梳理几个关键步骤和优化点，确保监控的精准度和实用性：

1. 自定义StreamingQueryListener的核心实现 #

首先要正确重写onQueryProgress方法，精准提取你关注的StreamingQueryProgress指标。这里要注意不同场景下的指标获取逻辑：

• numInputRecords：单数据源场景直接取progress.numInputRecords；多数据源合并的话，要遍历progress.sources数组，累加每个SourceProgress对象里的对应值，避免数据统计遗漏
• inputRowsPerSecond：单数据源直接用progress.inputRowsPerSecond，多数据源建议分别记录每个源的速率，方便定位数据流入瓶颈
• processedRowsPerSecond：这是整个微批的全局处理速率，直接从progress.processedRowsPerSecond获取即可
• triggerExecution：注意这个值是Optional[Long]类型，要处理空值情况，比如用progress.durationMs.getOrElse("triggerExecution", 0L)避免空指针异常

给你一个可复用的Listener示例（Scala版本）：

import org.apache.spark.sql.streaming.{StreamingQueryListener, StreamingQueryProgress}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import java.util.Properties
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule

class KafkaMetricsListener(kafkaTopic: String, kafkaProps: Properties) extends StreamingQueryListener {
  private val producer = new KafkaProducer[String, String](kafkaProps)
  private val mapper = new ObjectMapper().registerModule(DefaultScalaModule)

  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {}

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    val progress = event.progress
    // 构造带维度的监控指标
    val metricsMap = Map(
      "queryId" -> progress.id.toString,
      "batchId" -> progress.batchId.toString,
      "timestamp" -> progress.timestamp,
      "numInputRecords" -> progress.numInputRecords.toString,
      "inputRowsPerSecond" -> progress.inputRowsPerSecond.toString,
      "processedRowsPerSecond" -> progress.processedRowsPerSecond.toString,
      "triggerExecutionMs" -> progress.durationMs.getOrElse("triggerExecution", 0L).toString,
      "jobName" -> progress.name
    )
    // 用Jackson序列化避免字符串拼接的格式错误
    val metricsJson = mapper.writeValueAsString(metricsMap)
    // 异步发送到Kafka，避免阻塞微批处理
    producer.send(new ProducerRecord[String, String](kafkaTopic, metricsJson), (metadata, exception) => {
      if (exception != null) {
        println(s"Failed to send metrics: ${exception.getMessage}")
      }
    })
  }

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
    producer.close()
  }
}

2. 提升指标采集精准性的关键细节 #

• 处理故障恢复场景：开启checkpoint后，作业重启可能会重跑部分微批，建议把batchId加入指标作为唯一标识，在Kafka消费端做去重，避免重复统计
• 补充端到端延迟指标：除了你提到的几个指标，建议加上progress.delayMs（数据从生成到被处理的延迟），这对监控流作业的实时性至关重要
• 强类型序列化：尽量用Jackson这类JSON序列化库代替字符串拼接，避免因类型转换或格式错误导致监控数据失效

3. 正确注册Listener到SparkSession #

要确保Listener在流作业启动前注册，否则会错过初始批次的监控数据：

val spark = SparkSession.builder()
  .appName("StructuredStreamPerformanceMonitor")
  .getOrCreate()

// 配置Kafka生产者属性
val kafkaProps = new Properties()
kafkaProps.put("bootstrap.servers", "your-kafka-broker:9092")
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

// 注册自定义Listener
spark.streams.addListener(new KafkaMetricsListener("spark-stream-metrics-topic", kafkaProps))

// 启动你的结构化流作业
val query = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:9092")
  .option("subscribe", "input-topic")
  .load()
  // 这里替换成你的业务处理逻辑
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:9092")
  .option("topic", "output-topic")
  .start()

query.awaitTermination()

4. Kafka端的监控闭环处理 #

• 指标持久化：用Kafka Consumer把监控数据写入时序数据库（比如InfluxDB、Prometheus），方便存储和历史趋势分析
• 可视化告警：用Grafana对接时序数据库，创建专属仪表盘，展示：
  • numInputRecords的批次波动趋势
  • inputRowsPerSecond与processedRowsPerSecond的速率对比（快速定位处理瓶颈）
  • triggerExecutionMs的变化曲线（监控微批处理耗时）
    还可以设置告警规则，比如当处理速率远低于流入速率时，自动触发通知

5. 进阶优化建议 #

• 异步发送指标：示例中用了Kafka的异步发送回调，避免阻塞流作业的微批处理逻辑
• 添加多维度标签：比如在指标中加入部署环境（dev/prod）、数据源类型等标签，方便在监控平台做多维度聚合分析
• 监控作业状态：在onQueryTerminated方法中记录作业终止原因，及时发现异常退出的情况

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