这些是使用 Confluent 平台序列化的 Avros。
我想找到一个像这样的工作示例:
但对于 Spark 结构化流。
kafka
.select("value")
.map { row =>
// this gives me test == testRehydrated
val test = Foo("bar")
val testBytes = AvroWriter[Foo].toBytes(test)
val testRehydrated = AvroReader[Foo].fromBytes(testBytes)
// this yields mangled Foo data
val bytes = row.getAs[Array[Byte]]("value")
val rehydrated = AvroReader[Foo].fromBytes(bytes)
最佳答案
我们一直在研究这个可能有帮助的库:ABRiS (Avro Bridge for Spark)
它提供 API,用于在读取和写入操作(流式处理和批处理)中将 Spark 集成到 Avro。它还支持 Confluent Kafka 并与 Schema Registry 集成。
免责声明:我为 ABSA 工作,我是这个库背后的主要开发人员。
关于scala - 如何使用 Spark Streaming 从 Kafka 读取二进制序列化的 Avro(Confluent Platform),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41089069/