本文主要从以下几个方面介绍SparkSQL中的DataFrame:

第一，SparkSQL的作用

第二，什么是DataFrame

第三，DataFrame与RDD的区别

第四，DataFrame的创建与使用 （Spark1.x与Spark2.x两种不同版本）

第一，SparkSQL的作用

SparkSQL是spark处理结构化数据的一个模块，它的前身是shark，与基础的spark RDD不同，SparkSQL提供了结构化数据及计算结果等信息的接口，在内部，SparkSQL使用这个额外的信息去执行额外的优化，有几种方式可以跟SparkSQL进行交互，包括SQL和DataSet API，使用相同的执行引擎进行计算的时候，无论是使用哪一种计算引擎都可以快速的计算。

在使用RDD进行处理时，需要了解RDD的每个算子的特点，以求得高效的执行相应的操作。在使用SparkSQL时，程序会自动优化算子的执行过程以及使用哪个算子，提高效率。SparkSQL主要用于进行结构化数据的处理，作为分布式的SQL查询引擎。

第二，什么是DataFrame

DataFrame的前身是SchemaRDD，从Spark 1.3.0开始SchemaRDD更名为DataFrame。与SchemaRDD的主要区别是：DataFrame不再直接继承自RDD，而是自己实现了RDD的绝大多数功能。你仍旧可以在DataFrame上调用rdd方法将其转换为一个RDD。DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库的二维表格，DataFrame带有Schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。

1.DataFrame的作用：

是Spark SQL提供的最核心的编程抽象。

以列的形式组织的，分布式的数据集合。

它其实和关系型数据库中的表非常类似，但是底层做了很多的优化。

2.DataFrame可以构建的来源：

（1）结构化的数据文件

（2）Hive中的表

（3）外部的关系型数据库

（4）RDD

第三，DataFrame与RDD的区别

RDD可看作是分布式的对象的集合，Spark并不知道对象的详细模式信息，DataFrame可看作是分布式的Row对象的集合，其提供了由列组成的详细模式信息，使得Spark SQL可以进行某些形式的执行优化。DataFrame和普通的RDD的逻辑框架区别如下所示：

左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解 Person类的内部结构。

而右侧的DataFrame却提供了详细的结构信息，使得Spark SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。DataFrame多了数据的结构信息，即schema。这样看起来就像一张表了，DataFrame还配套了新的操作数据的方法，DataFrame API（如df.select())和SQL(select id, name from xx_table where ...)。

此外DataFrame还引入了off-heap,意味着JVM堆以外的内存, 这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作。

RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化。

有了DataFrame这个高一层的抽象后，我们处理数据更加简单了，甚至可以用SQL来处理数据了，对开发者来说，易用性有了很大的提升。

不仅如此，通过DataFrame API或SQL处理数据，会自动经过Spark 优化器（Catalyst）的优化，即使你写的程序或SQL不高效，也可以运行的很快。

1、DataFrame与RDD的优缺点

RDD的优缺点：

优点:

（1）编译时类型安全

编译时就能检查出类型错误

（2）面向对象的编程风格

直接通过对象调用方法的形式来操作数据

缺点:

（1）序列化和反序列化的性能开销

无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化。

关于序列化问题可以参考这篇文章：
https://www.toutiao.com/i6846761221337809419/

（2）GC的性能开销

频繁的创建和销毁对象, 势必会增加GC

DataFrame通过引入schema和off-heap（不在堆里面的内存，指的是除了不在堆的内存，使用操作系统上的内存），解决了RDD的缺点, Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了；通过off-heap引入，可以快速的操作数据，避免大量的GC。但是却丢了RDD的优点，DataFrame不是类型安全的, API也不是面向对象风格的。

第四，DataFrame的创建与使用

spark1.x版本

在该版本下，我选择使用的scala2.10.7版本

spark1.x版本中总共有三种编程模式：

1、RDD数据类型调用toDF方法将RDD转化为DataFrame

package xxx

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

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联case class，将非结构化数据转化为结构化数据
 * 3、显示的调用toDF方法将RDD转化为DataFrame
 * 4、注册临时表
 * 5、执行SQL（Transformation方式）
 * 6、执行action
 */
object SqlDemo1 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val boyRDD: RDD[Person] = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Person(id, name, age, score)
    })

    //该RDD装的是Person类型的数据，有了shcma信息，但是还是一个RDD
    //将RDD转换成DataFrame
    //导入隐式转换
    import sqlContext.implicits._
    val bdf: DataFrame = boyRDD.toDF

    //变成DF后就可以使用两种API进行编程了
    //把DataFrame先注册临时表
    bdf.registerTempTable("t_boy")

    //书写SQL（SQL方法应其实是Transformation）
    val result: DataFrame = sqlContext.sql("SELECT * FROM t_boy ORDER BY score desc, age asc")

    //查看结果（触发Action）
    result.show()

    sc.stop()

  }

  case class Person(id: Int, name: String, age: Int, score: Int)
}

2、在RDD中关联Row，将非结构化的数据转化为结构化的数据，然后定义模式，通过调用SqlContext的createDataFrame方法将RDD转化为DataFrame

package xxx

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StructField, StructType, StringType}
import org.apache.spark.sql.{DataFrame, Row, SQLContext, types}

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用sqlContext的createDataFrame方法
 * 5、注册临时表
 * 6、执行SQL（Transformation方式）
 * 7、执行action
 */
object SqlDemo2 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Row(id, name, age, score)
    })

    // 设置结构类型，表头信息
    val structType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", IntegerType, true)
    ))

    // 将RowRDD关联schema
    val bdf = sqlContext.createDataFrame(rowRDD, structType)

    //变成DF后就可以使用两种API进行编程了
    //把DataFrame先注册临时表
    bdf.registerTempTable("t_boy")

    //书写SQL（SQL方法应其实是Transformation）
    val result: DataFrame = sqlContext.sql("SELECT * FROM t_boy ORDER BY score desc, age asc")

    //查看结果（触发Action）
    result.show()

    sc.stop()

  }

}

3、创建DataFrame的方式同1，只是在使用DataFrame的时候不使用SQL语句，而是使用DataFrame API，这样就不需要将DataFrame注册成临时表

package xxx

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用sqlContext的createDataFrame方法
 * 6、执行DataFrame语句（Transformation方式）
 * 7、执行action
 */
object SqlDemo3 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Row(id, name, age, score)
    })

    // 设置结构类型，表头信息
    val structType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", IntegerType, true)
    ))

    // 将RowRDD关联schema
    val bdf = sqlContext.createDataFrame(rowRDD, structType)

    // 不使用SQL的方式，就不需要注册临时表
    val frame = bdf.select("id", "name", "score")


    import  sqlContext.implicits._
    val frame1 = bdf.orderBy(#34;score" desc, #34;age" asc)

    frame1.show()

    sc.stop()
  }

}

在Spark2.x版本中，引入了SparkSession的概念，它为用户提供了一个统一的切入点来使用Spark的各项功能，用户不但可以使用DataFrame的各种API和可以使用SQL，学习Spark2的难度也会大大降低。SparkConf、SparkContext和SQLContext都已经被封装在SparkSession当中。

在该版本下，我选择使用的scala2.11.12版本

1、创建SparkSession,由于SparkContext被封装在SparkSession中，直接调用
SparkSession.SparkContext.textFile读取文件,在RDD中关联Row，将非结构化的数据转化为结构化的数据，然后定义模式，通过调用SqlContext的createDataFrame方法将RDD转化为DataFrame.

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * Spark2.x 编程模式
 * 1、创建SparkSession,SparkContext被封装在SparkSession中，直接调用读取文件
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用SparkSession的createDataFrame方法
 * 5、注册临时表
 * 6、执行DataFrame语句（Transformation方式）
 * 7、执行action
 */
object SparkTest1 {
  def main(args: Array[String]): Unit = {

    //spark2.x SQL的编程API(SparkSession)
    //是spark2.x SQL执行的入口
    val session = SparkSession.builder()
      .appName("SQLTest1")
      .master("local[*]")
      .getOrCreate()

    //创建RDD
    val lines: RDD[String] = session.sparkContext.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD: RDD[Row] = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toDouble
      Row(id, name, age, score)
    })

    //结果类型，其实就是表头，用于描述DataFrame
    val schema: StructType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", DoubleType, true)
    ))

    //创建DataFrame
    val df: DataFrame = session.createDataFrame(rowRDD, schema)

    // 导入隐式转化
    import session.implicits._
    val df2: Dataset[Row] = df.where(#34;score" > 98).orderBy(#34;score" desc, #34;age" asc)

    df2.show()

    session.stop()
  }

}

2、创建DataFrame后，同样也可以调用createTempView将DataFrame注册成视图，在视图里使用SQL进行查询。如上面的一样，省略。。。。