使用Spark结构化流媒体编写时捕获Kafka偏移量

Question

我正在使用Spark 2.2上的Spark结构化流媒体将文件从HDFS目录流式传输到Kafka主题。我想为我写的主题数据捕捉Kafka偏移量。

我正在使用

val write = jsonDF
.writeStream.format("kafka")
.option("checkpointLocation", Config().getString(domain + ".kafkaCheckpoint"))
.option("kafka.bootstrap.servers", Config().getString(domain + ".kafkaServer"))
.option("topic", Config().getString(domain + ".kafkaTopic"))
.start()

给Kafka写信。

当我利用

spark.streams.addListener(new StreamingQueryListener() {
  override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
    println("Query started: " + queryStarted.id) 
  }
  override def onQueryTerminated(queryTerminated: QueryTerminatedEvent): Unit = {
    println("Query terminated: " + queryTerminated.id)
  }
  override def onQueryProgress(queryProgress: QueryProgressEvent): Unit = {
    println("Query made progress: " + queryProgress.progress)
  }
})

为了捕获流的进度信息，检索到的信息与Kafka中创建的偏移量不相关。

我假设这是因为流提供的信息实际上是关于我正在使用的文件流的，而与Kafka中编写的内容无关。

有没有一种Spark Structure流式处理方法可以捕获写入Kafka时生成的偏移量信息？

添加示例：当我在创建主题后从源1运行三行数据时，我得到：
运行1：开始偏移量：null，结束偏移量：{“logOffset”：0}开始偏移量：{“logOffset”：0}，结束偏移量：{“logOffset”：0}

 Kafka Says:
 ruwe:2:1
 ruwe:1:1
 ruwe:0:1

运行2；

  Start Offset: {"logOffset":0}, End offset: {"logOffset":1}
  Start Offset: {"logOffset":1}, End offset: {"logOffset":1}

 Kafka Says:
 ruwe:2:2
 ruwe:1:2
 ruwe:0:2

运行3：

  Start Offset: {"logOffset":1}, End offset: {"logOffset":2}
  Start Offset: {"logOffset":2}, End offset: {"logOffset":2}

 Kafka Says:
 ruwe:2:3
 ruwe:1:3
 ruwe:0:3

然后我用来自不同来源的相同程序运行数据并接收

  Start Offset: null, End offset: {"logOffset":0}
  Start Offset: {"logOffset":0}, End offset: {"logOffset":0}

  and of course Kafka continued to increment

这表示Spark正在报告基于源的信息

我想知道目标中创建了什么。

Answer 1

在阅读Spark结构流的代码后，特别是Kafka KafkaWriter、KafkaWriteTask和CachedKafkaProducer，Spark不会消耗回调中从KafkaProducer返回的偏移量。他们定义的回调只捕获异常。基于此，我想说在当前版本2.2中它无法完成。

他们提供的信息都是围绕查询源而不是目标提供的。

Answer 2

有没有一种Spark Structure流式处理方法可以捕获写入Kafka时生成的偏移量信息？

是的，在onQuerygres中，您需要查看StreamingQueryProgress.sources，它是一个Array[Sourcegres]。它有两个字符串，start Offset和endOffset，它们是您可以解析的JSON：

sparkSession.streams.addListener(new StreamingQueryListener {override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = ???

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    val source = event.progress.sources.headOption
    source.map(src => println(s"Start Offset: ${src.startOffset}, End offset: ${src.endOffset}"))
  }

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

JSON具有以下结构：

"startOffset" : {
  "topic-name" : {
    "0" : 1,
    "1" : 22,
    "2" : 419,
  }
},
"endOffset" : {
  "topic-name" : {
    "0" : 10,
    "1" : 100,
    "2" : 1000
  }
}