1.什么是 Spark SQL

Spark SQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用。

2. DataFrame 详解

与RDD类似，DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema。同时，与Hive类似，DataFrame也支持嵌套数据类型（struct、array和map）。从API易用性的角度上 看，DataFrame API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。由于与R和Pandas的DataFrame类似，Spark DataFrame很好地继承了传统单机数据分析的开发体验。

DataFrame

如何创建 DataFrame

在Spark SQL中SQLContext是创建DataFrames和执行SQL的入口，在spark-1.5.2中已经内置了一个sqlContext

步骤：

1. 在本地创建一个文件，有三列，分别是id、name、age，用空格分隔，然后上传到hdfs上

    hdfs dfs -put person.txt /

2. 在spark shell执行下面命令，读取数据，将每一行的数据使用列分隔符分割

    val lineRDD = sc.textFile("hdfs://node1.itcast.cn:9000/person.txt").map(_.split(" "))

3. 定义case class（相当于表的schema）

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

4. 将RDD和case class关联

    val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))

5. 将RDD转换成DataFrame

  val personDF = personRDD.toDF

6. 对DataFrame进行处理

  personDF.show

scala> val rdd1 = sc.textFile("E:\\spark\\sql\\person.txt").map(_.split(","))
rdd1: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[2] at map at <console>:24

scala> case class Person(name: String, size: Int, age: Int)
defined class Person

scala> val prdd = rdd1.map(x => Person(x(0), x(1).toInt, x(2).toInt))
prdd: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[3] at map at <console>:28

scala> val df = prdd.toDF
df: org.apache.spark.sql.DataFrame = [name: string, size: int ... 1 more field]

scala> df.show
+---------+----+---+
|     name|size|age|
+---------+----+---+
|    julia| 100| 18|
|   liuyan|  98| 23|
|snowwhite|  98| 20|
+---------+----+---+

SQL 风格语法

如果想使用 SQL 风格的语法，需要将 DataFrame 注册成临时表。personDF.registerTempTable("t_person")

查询实例：

sqlContext.sql("select * from t_person order by age desc limit 2")

3. Spark SQL 编程

3.1 通过反射推断 schema

package cn.itcast.spark.sql

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

object SqlDemo {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SqlDemo").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._
    val df = sc.textFile("E:/spark/sql/person.txt").map(line => {
      val fields = line.split(",")
      Person(fields(0), fields(1).toInt, fields(2).toInt)
    }).toDF

    df.registerTempTable("t_person")

    val result = sqlContext.sql("select * from t_person order by size desc, age asc")
    result.show()
    
    result.write.json("E:/spark/sql/json")// 将结果以json格式写出去
    sc.stop()
  }
}

case class Person(name: String, size: Int, age: Int) extends Serializable

3.2 通过StructType直接指定Schema

package cn.itcast.spark.sql

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

object SpecifyingSchema {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SpecifyingSchema").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    val girlRdd = sc.textFile("E:/spark/sql/person.txt").map(_.split(","))
    //通过StructType直接指定每个字段的schema
    val schema = StructType(List(StructField("name", DataTypes.StringType, true)
      , StructField("size", DataTypes.IntegerType, true)
    , StructField("age", DataTypes.IntegerType, true)))

    val rowRdd = girlRdd.map(g => Row(g(0).trim, g(1).toInt, g(2).toInt))

    val df = sqlContext.createDataFrame(rowRdd, schema)
    df.registerTempTable("girl")

    val res = sqlContext.sql("select * from girl")
    res.show()

    sc.stop()

  }
}

4. Spark-SQL 操作 JDBC（待整理）