Отображение количества событий, использованных заданием структурированной потоковой передачи Spark

Question

У меня есть задание Spark Streaming с двумя вкладышами, которое копирует данные из одной kafka topi c в другую.

Можно ли публиковать / просматривать количество событий, использованных / произведенных в пользовательском интерфейсе Spark?

Answer 1

«Вкладка потоковой передачи» в веб-интерфейсе Spark недоступна для структурированной потоковой передачи, только для Direct API. Однако есть еще один простой способ отображения количества событий, обработанных заданием Spark Structured Streaming.

Вы можете использовать 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 содержит дополнительную информацию, которая поможет вам понять данные обработка вашего потокового задания.