这个问题在这里已经有答案了：

集成spark结构化流与融合模式注册表（7个答案）
两年前关门了。
我正在研究spark流上下文，它从avro序列化中的kafka主题获取数据，如下所示。

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092",
  "schema.registry.url" -> "http://localhost:8081",
  "key.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer",
  "value.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer",
  "group.id" -> "1"
)

使用kafka utils，我创建了如下的直接流

val topics = Set("mysql-foobar")

val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String,String](
    topics,
    kafkaParams)
)

我还将数据作为

stream.foreachRDD ( rdd => {
  rdd.foreachPartition(iterator => {
    while (iterator.hasNext) {
      val next = iterator.next()
      println(next.value())
    }
  })
})

现在我想从这些rdd创建Dataframe。有没有可能我已经审查和测试了许多解决方案，从stackoverflow，但得到一些问题与他们。stackoverflow的解决方案是this和this。我的输出如下

{"c1": 4, "c2": "Jarry", "create_ts": 1536758512000, "update_ts": 1537204805000}