二、建立DataFrame的方式

#1：DataFrame其实它是经过RDD的map方法读取每一条数据，而后把他存到“case class”中。最后经过这个RDD的toDF方法产生的。 #因为case class 中有属性字段，并且这些字段的类型都有了，是否是就是表结构？？

#2:能够经过sqlContext.read.format("jdbc").options(）建立DataFrame（详情笔记三）

#如何建立？代码以下

package com.liufu.org.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by liufu on 2016/11/18.
  */
object DataFrameTest {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("dataframTest").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    //建立SQLContext
    val sqlContext = new SQLContext(sc)

    //读取外部数据，而且切割成每一条信息,造成arr数组。
    val spilted: RDD[Array[String]] = sc.textFile("file:///E:/words.txt").map(_.split(" "))

    //这里的操做很容易出错，
    //当map中的逻辑很复杂的时候，通常不建议使用“_”来代替每个数据，
    // 而是使用“函数体”的形式来一步一步的操做实现。
    val personRDD: RDD[Person] = spilted.map(valueArr => Person(valueArr(0),valueArr(1).toInt))

    /**
      * 导入隐式转换，若是不到人没法将RDD转换成DataFrame
      * 将上述有RDD和case class产生的personSet变化成DataFrame。
      */
    import sqlContext.implicits._
    val dF: DataFrame = personRDD.toDF()

    //将DataFrame注册成表结构才可以使用SQL进行操做。
    dF.registerTempTable("person")

    //使用sqlContext执行sql语句进行查询操做。
    val selectDF: DataFrame = sqlContext.sql("select * from person")
    selectDF.show()

    sc.stop()
  }
}

//经过样例类来定义表结构。还能够经过structType产生一个shema的方式，可是太麻烦通常不怎么用
case class Person(name:String,age:Int)

#总结： 使用DataFrame的流程

• 1：RDD的map方法中将信息写入到Person样例类中，这样数据就和表结构关联起来了。

  val personRDD: RDD[Person] = spilted.map(valueArr => Person(valueArr(0),valueArr(1).toInt))

• 2：将RDD转化为DataFrame。注意：必定要引入sqlContext的隐式转换。

  /**
      * 导入隐式转换，若是不到人没法将RDD转换成DataFrame
      * 将上述有RDD和case class产生的personSet变化成DataFrame。
      */
    import sqlContext.implicits._
    val dF: DataFrame = personRDD.toDF()

• 3：将DataFrame注册成表，由于只有注册成表以后才可以使用sql语句来查询

  dF.registerTempTable("person")

• 4：利用sqlContext对象来执行SQL语句来对表进行操做。

  val selectDF: DataFrame = sqlContext.sql("select * from person")
    //展现。 
    selectDF.show()

• 5：//将结果以JSON的方式存储到指定位置

  df.write.json("file:///E:/output1")