错误:值字符串不是org.apache.spark.rdd.RDD[org.apache.kafka.clients.consumer.ConsumerRecord[String，toDF的成员]]

Question

我正在尝试使用Scala中的sparkStreaming来捕获Kafka事件(我正在以序列化的形式获取它)。

下面是我的代码片段：

val spark = SparkSession.builder().master("local[*]").appName("Spark-Kafka-Integration").getOrCreate()
spark.conf.set("spark.driver.allowMultipleContexts", "true")

val sc = spark.sparkContext
val ssc = new StreamingContext(sc, Seconds(5))

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._

val topics=Set("<topic-name>")
val brokers="<some-list>"
val groupId="spark-streaming-test"

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> brokers,
  "auto.offset.reset" -> "earliest",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "group.id" -> groupId,
  "enable.auto.commit" -> (false: java.lang.Boolean)
)

val messages: InputDStream[ConsumerRecord[String, String]] =
  KafkaUtils.createDirectStream[String, String](
    ssc,
    LocationStrategies.PreferConsistent,
    ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
  )

messages.foreachRDD { rdd =>
  println(rdd.toDF())
}

ssc.start()
ssc.awaitTermination()

我收到错误消息: Error:(59，19) value toDF不是org.apache.spark.rdd.RDD[org.apache.kafka.clients.consumer.ConsumerRecordString，字符串的成员]toDF(rdd.toDF())

Answer 1

toDF通过DatasetHolder实现

https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.SQLImplicits

我没有复制它，但我猜没有用于ConsumerRecord[String, String]的编码器，因此您可以提供一个编码器，或者首先将其映射到可以派生Encoder的对象(case类或原语)。

另外，由于spark的分布式特性，foreachRDD中的println可能不会按照您想要的方式运行