我正在使用 Spark 2.0.2 和 Kafka 0.11.0,并且
我正在尝试在火花流中使用来自卡夫卡的消息。以下是代码:
val topics = "notes"
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "localhost:7092",
"schema.registry.url" -> "http://localhost:7070",
"group.id" -> "connect-cluster1",
"value.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer",
"key.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer"
)
val topicSet: Set[String] = Set(topics)
val stream = KafkaUtils.createDirectStream[String, String](
SparkStream.ssc,
PreferConsistent,
Subscribe[String, String](topicSet, kafkaParams)
)
stream.foreachRDD ( rdd => {
rdd.foreachPartition(iterator => {
while (iterator.hasNext) {
val next = iterator.next()
println(next.value())
}
})
})
如果 Kafka 消息包含记录,则输出将为:
{"id": "4164a489-a0bb-4ea1-a259-b4e2a4519eee", "createdat": 1505312886984, "createdby": "karthik", "notes": "testing20"}
{"id": "4164a489-a0bb-4ea1-a259-b4e2a4519eee", "createdat": 1505312890472, "createdby": "karthik", "notes": "testing21"}
因此,从 ConsumerRecord 的值可以看出,接收到的消息是 Avro 解码的。
现在我需要数据帧格式的这些记录,但我不知道如何从这里继续,即使手头的模式如下:
val sr : CachedSchemaRegistryClient = new CachedSchemaRegistryClient("http://localhost:7070", 1000)
val m = sr.getLatestSchemaMetadata(topics + "-value")
val schemaId = m.getId
val schemaString = m.getSchema
val schemaRegistry : CachedSchemaRegistryClient = new CachedSchemaRegistryClient("http://localhost:7070", 1000)
val decoder: KafkaAvroDecoder = new KafkaAvroDecoder(schemaRegistry)
val parser = new Schema.Parser()
val avroSchema = parser.parse(schemaString)
println(avroSchema)
打印的模式如下:
{"type":"record","name":"notes","namespace":"db","fields":[{"name":"id","type":["null","string"],"default":null},{"name":"createdat","type":["null",{"type":"long","connect.version":1,"connect.name":"org.apache.kafka.connect.data.Timestamp","logicalType":"timestamp-millis"}],"default":null},{"name":"createdby","type":["null","string"],"default":null},{"name":"notes","type":["null","string"],"default":null}],"connect.name":"db.notes"}
谁能帮助我了解如何从消费者记录的值中获取数据框?我看过其他问题,例如使用 schema 将 Spark 的 AVRO 消息转换为 DataFrame https://stackoverflow.com/questions/39049648/use-schema-to-convert-avro-messages-with-spark-to-dataframe/39072520, ,但他们一开始并没有处理consumerRecord。