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就行了，报的数组下标越界，懵了半天。