Apache Kafka with Structured Streaming protobuf

Question

我正在尝试使用结构化流编写一个 Kafka 消费者(protobuf(。让我们称protobuf为A，在Scala中应该反序列化为字节数组(Array[Byte](。

我尝试了我能在网上找到的所有方法，但仍然无法弄清楚如何正确解析消息A

方法 1：从集成指南 (https://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html( 中，我应该将值转换为字符串。但是即使我确实getBytes将字符串转换为字节以解析我的消息A，我也得到：

Exception in thread "main" java.lang.ExceptionInInitializerError
...
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Incompatible Jackson version: 2.9.8

方法2：我想将值直接转换为字节数组。我会得到：

missing ')' at '['(line 1, pos 17)
== SQL ==
CAST(key AS Array[Byte])

方法3：我按照(https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html(编写了我自己的protobuf反序列化器。但收到错误消息：

Schema for type A is not supported

以上三种方法大概是我能在网上找到的所有方法。这应该是一个简单而常见的问题，所以如果有人对此有见解，请告诉我。

谢谢！

Answer 1

从流式处理源创建的DataFrame的架构为：

root
|-- key: binary (nullable = true)
|-- value: binary (nullable = true)
|-- topic: string (nullable = true)
|-- partition: integer (nullable = true)
|-- offset: long (nullable = true)
|-- timestamp: timestamp (nullable = true)
|-- timestampType: integer (nullable = true)

所以键和值实际上是Array[Byte].您必须在数据帧操作中执行反序列化。

例如，我有这个用于卡夫卡反序列化：

import sparkSession.implicits._
sparkSession.readStream
.format("kafka")
.option("subscribe", topic)
.option(
"kafka.bootstrap.servers",
bootstrapServers
)
.load()
.selectExpr("key", "value") // Selecting only key & value
.as[(Array[Byte], Array[Byte])]
.flatMap {
case (key, value) =>
for {
deserializedKey <- Try {
keyDeserializer.deserialize(topic, key)
}.toOption
deserializedValue <- Try {
valueDeserializer.deserialize(topic, value)
}.toOption
} yield (deserializedKey, deserializedValue)
}

您需要修改它以反序列化您的 protobuf 记录。