在 spark-kafka 中使用模式将 ConsumerRecord 值转换为 Dataframe答案

【问题标题】：Use schema to convert ConsumerRecord value to Dataframe in spark-kafka在 spark-kafka 中使用模式将 ConsumerRecord 值转换为 Dataframe
【发布时间】：2018-02-22 08:13:43
【问题描述】：

我正在使用 Spark 2.0.2 和 Kafka 0.11.0，以及我正在尝试在火花流中使用来自 kafka 的消息。以下是代码：

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"}

谁能帮助我了解如何从消费者记录的价值中获取数据框？我已经查看了其他问题，例如Use schema to convert AVRO messages with Spark to DataFrame、Handling schema changes in running Spark Streaming application，但它们并没有首先处理 consumerRecord。

【问题讨论】：

我遇到了类似的情况。你能解决这个问题吗？

标签： scala apache-spark apache-kafka

【解决方案1】：

您可以在下面使用 sn-p ： stream是kafka010的kafkaUtils api返回的消费者记录的DStream：

stream.foreachRDD(rdd =>
    if (!rdd.isEmpty()) {
        val sqlContext = SQLContext.getOrCreate(rdd.sparkContext)
        import sqlContext.implicits._
        val topicValueStrings = rdd.map(record => (record.value()).toString)
        val df = sqlContext.read.json(topicValueStrings)
        df.show()
    })

【讨论】：

【解决方案2】：

我自己是 scala\kafka\spark 的新手，所以我不确定这是否能准确回答这个问题，但它会帮助我。我相信有比这更好的方法，所以希望有更多经验的人能提供更好的答案。

// KafkaRDD
stream.foreachRDD { rdd => {

  // pull the values I'm looking for into a string array
  var x = rdd.map(row => row.value()).collect()

  // convert to dataframe
  val df = spark.createDataFrame(x).toDF("record")

  // write data frame to datastore (MySQL in my case)
  df.write
    .mode(SaveMode.Append)
    .jdbc(url, table, props)

  }
}

【讨论】：