结构化流式传输 Kafka 源偏移存储

Question

我正在使用 Kafka 的结构化流源 (Integration guide)，如前所述，它不会提交任何偏移量。

我的目标之一是监控它（检查它是否落后等）。即使它不提交偏移量，它也会通过不时查询 kafka 并检查下一个要处理的偏移量来处理它们。根据文档，偏移量被写入 HDFS，因此在发生故障时可以恢复，但问题是：

它们存放在哪里？有没有什么方法可以监控火花流（结构化）的 kafka 消耗（从程序外部；所以 kafka cli 或类似的，每条记录附带的偏移量不适合用例）如果它不提交偏移量？

干杯

Answer 1

注意事项：

Monitoring：可以在 Spark 作业的 Streaming 选项卡中找到现成的监控。您可以查看当前正在处理的批次是什么，以及有多少批次在排队以检查延迟。

检查主题的最大和最小偏移量：您有 cli 来检查这些。可以从存在 kafka 代理的服务器使用以下内容：

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

如果集成Grafana，可以获得更详细的信息

Answer 2

kafka 的结构化流式处理将偏移量保存到结构下方的 HDFS。

检查点位置设置示例如下。

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

在这种情况下，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 主题。

这样，现有的偏移量滞后监控工具就可以监控 Structured Streaming 的 kafka 偏移量滞后。

Answer 3

方法一： 如果您配置了 checkpointLocation（HDFS/S3 等），请转到该路径，您会发现两个目录 offsets 和 commits。偏移量保存当前偏移量，而提交具有最后提交的偏移量。您可以导航到提交目录并打开最新修改的文件，您可以在其中找到最后提交的偏移量。而 offsets 目录中的最新文件包含消耗的偏移量信息。

方法二： 您还可以使用以下配置进行监控：

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())

结构化流式传输 Kafka 源偏移存储

Structured Streaming Kafka Source Offset Storage

offset

apache-kafka

apache-spark

spark-streaming

spark-structured-streaming