Spark Structured Streaming中StreamingQueryListener.onQueryProgress为何未按每个微批触发调用？

首先，我完全理解你的困惑——明明Spark UI显示每分钟都在生成新的微批作业，但onQueryProgress却隔5-6分钟才触发一次。这其实和Spark的进度报告机制有关：默认情况下，Spark会累积多个微批的进度事件后再批量推送给监听器，而非每个微批完成后立即调用。

下面是具体的排查和解决方法：

1. 调整进度报告间隔配置 #

Spark有一个关键配置spark.sql.streaming.reportProgressInterval，它控制着Spark向监听器发送进度更新的频率，默认值是10000毫秒（10秒）。如果你的环境中这个配置被修改为5-6分钟，或者因为某些原因导致事件累积，就会出现你遇到的情况。

你可以将这个配置设置为与触发时长一致（1分钟），确保每个微批的进度能及时被监听器捕获：

在SparkSession初始化时设置： #

val spark = SparkSession.builder()
  .appName("KafkaConsumerStream")
  .config("spark.sql.streaming.reportProgressInterval", "60000") // 60000毫秒=1分钟
  // 其他必要配置（如Kafka连接信息）
  .getOrCreate()

或者在提交作业时通过--conf参数指定： #

spark-submit \
  --conf spark.sql.streaming.reportProgressInterval=60000 \
  --class your.package.YourStreamingApp \
  your-streaming-jar.jar

2. 确认监听器注册正确 #

有时候问题可能出在监听器的注册环节，务必确保你已经正确将自定义的StreamingQueryListener添加到SparkSession中：

val customListener = new StreamingQueryListener() {
  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
    // 可选：处理作业启动事件
  }

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    // 这里会在每个微批进度报告触发时执行
    println(s"微批 ${event.progress.batchId} 完成，处理行数：${event.progress.numInputRows}")
  }

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
    // 可选：处理作业终止事件
  }
}

// 务必执行这一步添加监听器
spark.streams.addListener(customListener)

3. 排查空微批的优化逻辑 #

如果你的微批经常没有处理到Kafka数据（空微批），Spark可能会合并这些空微批的进度事件以减少开销。不过既然你已经在Spark UI中看到每分钟都有微批生成，这个因素的影响应该不大。调整上述的进度间隔配置后，即使是空微批也能及时触发进度报告。

4. 检查Driver资源 #

如果你的作业运行在集群模式下，Driver节点的CPU或内存不足可能导致进度事件处理延迟。可以查看Driver的监控指标（比如CPU使用率、堆内存使用情况），确保它有足够的资源及时处理进度事件。

调整完reportProgressInterval后，你应该会看到onQueryProgress的调用频率和触发时长保持一致了。

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