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

Question

创建如下所示的直接流后:

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

我想将上述流转换为数据帧，以便我可以对其运行配置单元查询。谁能解释一下这是如何实现的？我正在使用 spark 版本 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()
}