显示 Spark 结构化流作业消耗的事件数

Posted 2023-04-18

【中文标题】显示 Spark 结构化流作业消耗的事件数

【英文标题】：Display number of events consumed by a Spark Structured streaming job

【发布时间】：2020-04-14 21:45:57

【问题描述】：

我有 2 个线性 Spark Structured Streaming 作业，可将数据从一个 kafka 主题复制到另一个主题。

是否可以在 Spark UI 中发布/查看消耗/产生的事件数量？

【参考方案1】：

Spark Web UI 中的“流式处理选项卡”不适用于结构化流式处理，仅适用于 Direct API。从版本 3.x 开始可用。

但是，还有另一种简单的方法可以显示 Spark 结构化流作业处理的事件数量。

您可以使用StreamingQueryListener

import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener.QueryProgressEvent


class CountNumRecordsListener extends StreamingQueryListener 

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

  override def onQueryProgress(event: QueryProgressEvent): Unit = 
    println(s"""numInputRows: $event.progress.numInputRows""")
  

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit =  

使用该类，您可以将侦听器添加到您的流应用程序（其中spark 是您的SparkSession）。

val countNumRecordsListener = new CountNumRecordsListener
spark.streams.addListener(countNumRecordsListener)

StreamingQueryProgress 类提供了更多信息，可帮助您了解流作业的数据处理。