从Kafka流解析JSON RDD时的Java执行

Question

我正在尝试使用Spark stream库从kafka读取一个json字符串。该代码能够连接到kafka broker，但在解码消息时失败。代码的灵感来自

https://github.com/killrweather/killrweather/blob/master/killrweather-examples/src/main/scala/com/datastax/killrweather/kafkastreamingjson.scala

val kStream = KafkaUtils.createDirectStream[String, String, StringDecoder, 
         StringDecoder](ssc, kParams, kTopic).map(_._2)
  println("Starting to read from kafka topic:" + topicStr)
kStream.foreachRDD { rdd =>

   if (rdd.toLocalIterator.nonEmpty) {

          val sqlContext = new org.apache.spark.sql.SQLContext(sc)
            sqlContext.read.json(rdd).registerTempTable("mytable")
            if (firstTime) {
                sqlContext.sql("SELECT * FROM mytable").printSchema()
            }
            val df = sqlContext.sql(selectStr)
            df.collect.foreach(println)
            df.rdd.saveAsTextFile(fileName)
            mergeFiles(fileName, firstTime)
            firstTime = false
           println(rdd.name)
        }

Answer 1

问题出在使用的Kafka jars版本上，使用0.9.0.0修复了该问题。类Kafka.Message.MessageAndMetadata是在0.8.2.0中引入的。