如何在 Spark Streaming 中将 DStream[(Array[String], Long)] 转换为数据帧答案

【问题标题】：How to convert DStream[(Array[String], Long)] to dataframe in Spark Streaming如何在 Spark Streaming 中将 DStream[(Array[String], Long)] 转换为数据帧
【发布时间】：2019-07-03 18:00:36
【问题描述】：

我正在尝试将我的 dstream 转换为 Dataframe。这是用于将我的 dstream 转换为 Dataframe 的代码

           val ssc = new StreamingContext(spark.sparkContext, Seconds(10))
           val kafkaParams = Map[String, Object](
           "bootstrap.servers" -> "ffff.dl.uk.fff.com:8002",
           "security.protocol" -> "SASL_PLAINTEXT",
           "key.deserializer" -> classOf[StringDeserializer],
           "value.deserializer" -> classOf[StringDeserializer],
           "group.id" -> "1",
           "auto.offset.reset" -> "latest",
           "enable.auto.commit" -> (false: java.lang.Boolean)
           )

           val topics = Array("mytopic")
           val from_kafkastream = KafkaUtils.createDirectStream[String, 
           String](
           ssc,
           PreferConsistent,
           Subscribe[String, String](topics, kafkaParams)
           )
           val strmk = from_kafkastream.map(record => 
          (record.value,record.timestamp))
          val splitup2 = strmk.map{ case (line1, line2) => 
   (line1.split(","),line2)}

          case class Record(name: String, trQ: String, traW: String,traNS: 
   String, traned: String, tranS: String,transwer: String, trABN: 
  String,kafkatime: Long)

          object SQLContextSingleton {
            @transient  private var instance: SQLContext = _

            def getInstance(sparkContext: SparkContext): SQLContext = {
              if (instance == null) {
                instance = new SQLContext(sparkContext)
              }
              instance
            }
          }
          splitup2.foreachRDD((rdd) => {
          val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
          spark.sparkContext.setLogLevel("ERROR")
          import sqlContext.implicits._
          val requestsDataFrame = rdd.map(w => Record(w(0).toString, 
  w(1).toString, w(2).toString,w(3).toString, w(4).toString, 
  w(5).toString,w(6).toString, w(7).toString,w(8).toString)).toDF()
          // am getting issue here
          requestsDataFrame.show()
          })
          ssc.start()

我在关注时收到错误消息

有人可以帮助我如何将我的 dstreams 转换为 DF，因为我是新的 spark 世界

【问题讨论】：

一切似乎有点太复杂了。看看这个 git 示例 github.com/apache/spark/blob/master/examples/src/main/scala/org/….
这只是您提供的字数...您有解决方案吗？
不，因为我不会全部编译，但它告诉你这并不难。我觉得你在这里过于复杂了。如果您查看该解决方案或其他解决方案，它看起来都相当简单。这就是我的观点。我现在也看到了答案。

标签： spark-streaming

【解决方案1】：

也许错误是在构建 Record 对象时，因为您没有传递 kafkatime ，只有字符串值，而且是您无法访问此表单的属性数组的元组。

你可以试试这个：

import session.sqlContext.implicits._
val requestsDataFrame = rdd.map(w => Record(
  w._1(0).toString,
  w._1(1).toString, w._1(2).toString, w._1(3).toString, w._1.toString,
  w._1(5).toString, w._1(6).toString, w._1(7).toString, w._2))

requestsDataFrame.toDF()

【讨论】：

这个答案很有道理..让我试着找回你