如何将来自kafka的directstream转换为spark 1.3.0中的数据帧

Question

创建如下直接流后：

val events = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topicsSet)

我想将上面的流转换成数据帧，这样我就可以在它上面运行 hive 查询。谁能解释一下这是如何实现的？我使用的是火花版本 1.3.0

Answer 1

如Spark Streaming programming guide 中所述，试试这个：

import org.apache.spark.sql.SQLContext
object SQLContextSingleton {
  @transient private var instance: SQLContext = null

  // Instantiate SQLContext on demand
  def getInstance(sparkContext: SparkContext): SQLContext = synchronized {
    if (instance == null) {
      instance = new SQLContext(sparkContext)
    }
    instance
  }
}
case class Row(key: String, value: String)
eventss.foreachRDD { rdd =>
  val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
  import sqlContext.implicits._
  val dataFrame = rdd.map {case (key, value) => Row(key, value)}.toDF()
  dataFrame.show()
}