Структурированное потоковое хранилище Kafka Source Offset Storage

Структурированная потоковая передача для kafka сохраняет смещения в HDFS ниже структур.

Пример настройки checkpointLocation приведен ниже.

.writeStream.
.....
  option("checkpointLocation", "/tmp/checkPoint")
.....

В этом случае Structured Streaming для kafka сохраняет путь ниже

/tmp/checkPoint/offsets/$'batchid'

Сохраненный файл имеет следующий формат.

v1
{"batchWatermarkMs":0,"batchTimestampMs":$'timestamp',"conf":{"spark.sql.shuffle.partitions":"200"}}
{"Topic1WithPartiton1":{"0":$'OffsetforTopic1ForPartition0'},"Topic2WithPartiton2":{"1":$'OffsetforTopic2ForPartition1',"0":$'OffsetforTopic2ForPartition1'}}

Например.

v1
{"batchWatermarkMs":0,"batchTimestampMs":1505718000115,"conf":{"spark.sql.shuffle.partitions":"200"}}
{"Topic1WithPartiton1":{"0":21482917},"Topic2WithPartiton2":{"1":103557997,"0":103547910}}

Итак, я думаю, что для мониторинга задержки смещения необходимо разработать специальные инструменты, которые имеют следующие функции.

• Чтение по смещениям из HDFS.
• Напишите смещение в теме Kafka __offset.

Таким образом, уже существующий инструмент мониторинга задержки смещения может отслеживать структурированную потоковую передачу на предмет задержки смещения kafka.

Способ 1. Если вы настроили checkpointLocation (HDFS/S3 и т. д.), перейдите по пути и найдете два каталога offsets и commits. Смещения содержат текущие смещения, в то время как коммиты имеют последние зафиксированные смещения. Вы можете перейти в каталог коммитов и открыть последний измененный файл, в котором вы можете найти последние зафиксированные смещения. В то время как последний файл в каталоге смещений содержит информацию об использованных смещениях.

Метод 2. Вы также можете отслеживать то же самое со следующими конфигурациями:

class CustomStreamingQueryListener extends StreamingQueryListener with AppLogging {

  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
    logDebug(s"Started query with id : ${event.id}," +
      s" name: ${event.name},runId : ${event.runId}")
  }

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    val progress = event.progress
    logDebug(s"Streaming query made progress: ${progress.prettyJson}")
  }

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
    logDebug(s"Stream exited due to exception : ${event.exception},id : ${event.id}, " +
      s"runId: ${event.runId}")
  }

}

и добавьте его в конфигурацию ваших потоков.

spark.streams.addListener(new CustomStreamingQueryListener())

Несколько замечаний:

Мониторинг: готовый мониторинг можно найти на вкладке «Потоковая передача» задания Spark. Вы можете увидеть, какая текущая партия обрабатывается и сколько находится в очереди, чтобы проверить отставание.

Проверить максимальное и минимальное смещение для темы: у вас есть кли для проверки. Можно использовать ниже с сервера, на котором присутствует брокер kafka:

kafka-run-class \
kafka.tools.GetOffsetShell \
--broker-list your_broker1:port,your_broker2:port,your_broker3:port \
--topic your_topic \
--time -2

Более подробную информацию можно получить при интеграции с Grafana.