spark streaming向RDD和DataFrame轉換

Data streaming轉為DataFrame，不能直接一步轉到DF，需要先轉為RDD，然后再轉到DF，我們用流式處理數據后，再通過spark sql實時獲取我們想要的結果。

1.首先老規矩，創建spark上下文對象，spark SQL和spark Streaming，再創建個socket在Linux端打入數據。

val conf = new SparkConf().setAppName("Demo04DSToRDDToDF").setMaster("local[2]")

    conf.set("spark.sql.shuffle.partitions", "1")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    val ssc = new StreamingContext(sc,Durations.seconds(5))

    val lines = ssc.socketTextStream("master",8888)

2.首先用foreachRDD方法把spark streaming轉為RDD

    /**
      *
      * DS  -> RDD -> DF
      *
      */
    lines.foreachRDD(rdd => {
     val stuRDD = rdd.map(line => {
        val split = line.split(",")
       (split(0),split(1),split(2).toInt,split(3),split(4))
      })

3.導入sqlContext隱式轉換，將RDD To成DF，同時傳入column對應RDD返回值

      //導入隱式轉換，將RDD轉換為DF
      import sqlContext.implicits._

      val stuDF = stuRDD.toDF("id","name","age","gender","clazz")
      //注冊成表
      stuDF.registerTempTable("student")

      val result = sqlContext.sql("select clazz,count(1) from student group by clazz")

      result.show()
      result.write.mode(SaveMode.Append).json("Spark/data/dsondf")

總結，這里的sqlContext必須自己創建好，原來我還以為是導包的時候，類都已經封裝好了的，直接import就行了，報的數組下標越界，懵了半天。