Spark 学习（八） SparkSQL简介

一，Spark SQL概述

　　1.1 什么是Spark SQL

　　1.2 为什么学Spark SQL

二，DataFrames

　　2.1 什么是DataFrames

　　2.2 创建DataFrames

三，DataFrame常用操作

　　3.1 DSL风格语法

　　3.2 SQL风格语法

四，SparkSQL编程实例

　　4.1 前期准备

　　4.2 通过反射推断Schema

　　4.3 通过StructType直接指定Schema

　　4.4 操作DataFrameAPI的形式进行数据操作

　　  4.5 Spark2.X的sql实现方式

　　4.6 SparkDataSet

正文

一，Spark SQL概述

　　1.1 什么是Spark SQL

　　

　　

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

　　1.2 为什么学Spark SQL

　　我们已经学习了Hive，它是将Hive SQL转换成MapReduce然后提交到集群上执行，大大简化了编写MapReduce的程序的复杂性，由于MapReduce这种计算模型执行效率比较慢。所有Spark SQL的应运而生，它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快！

　　1.易整合

　　

　　2.统一的数据访问方式

　　

　　3.兼容Hive

　　

　　4.标准的数据连接

　　

二，DataFrames

　　2.1 什么是DataFrames

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

·　　

　　2.2 创建DataFrames

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

//2.在spark shell执行下面命令，读取数据，将每一行的数据使用列分隔符分割
val lineRDD = sc.textFile("hdfs://node1.xiaoniu.com: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

三，DataFrame常用操作

　　3.1 DSL风格语法

//查看DataFrame中的内容
personDF.show

//查看DataFrame部分列中的内容
personDF.select(personDF.col("name")).show
personDF.select(col("name"), col("age")).show
personDF.select("name").show

//打印DataFrame的Schema信息
personDF.printSchema

//查询所有的name和age，并将age+1
personDF.select(col("id"), col("name"), col("age") + 1).show
personDF.select(personDF("id"), personDF("name"), personDF("age") + 1).show


//过滤age大于等于18的
personDF.filter(col("age") >= 18).show


//按年龄进行分组并统计相同年龄的人数
personDF.groupBy("age").count().show()

　　3.2 SQL风格语法

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

//查询年龄最大的前两名
sqlContext.sql("select * from t_person order by age desc limit 2").show


//显示表的Schema信息
sqlContext.sql("desc t_person").show

四，SparkSQL编程实例

　　4.1 前期准备

　　前面我们学习了如何在Spark Shell中使用SQL完成查询，现在我们来实现在自定义的程序中编写Spark SQL查询程序。首先在maven项目的pom.xml中添加Spark SQL的依赖

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.11</artifactId>
      <version>${spark.version}</version>
</dependency>

　　4.2 通过反射推断Schema

　　创建一个object如下：

package cn.edu360.spark06

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

object spark1XDemo1 {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("spark1XDemo1").setMaster("local[2]")
        val sc = new SparkContext(conf)
        //将SparkContext包装进而增强
        val sQLContext = new SQLContext(sc)
        val lines: RDD[String] = sc.textFile("hdfs://hd1:9000/sparkSQL")
        val boyRDD: RDD[Boy] = lines.map(line => {
            val fields: Array[String] = line.split(",")
            val id: Long = fields(0).toLong
            val name: String = fields(1)
            val age: Int = fields(2).toInt
            val fv: Double = fields(3).toDouble
            Boy(id, name, age, fv)
        }
        )
        //该RDD装的是Boy类型的数据，有了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 fv desc, age asc")

        //查看结果（触发Action）
        result.show()
        sc.stop()
    }
}
// 若要将这些字段解析后进行结构化的描述信息
// 这里进行表的描述
case class Boy(id: Long, name: String, age: Int, fv: Double)

　　4.3 通过StructType直接指定Schema

　　创建一个object为：

package cn.edu360.spark06

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

object spark1XDemo2 {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("spark1XDemo2").setMaster("local[2]")
        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)
        val lines: RDD[String] = sc.textFile("hdfs://hd1:9000/sparkSQL/")

        //将数据进行整理
        val rowRDD: RDD[Row] = lines.map(line => {
            val fields: Array[String] = line.split(",")
            val id: Long = fields(0).toLong
            val name: String = fields(1)
            val age: Int = fields(2).toInt
            val fv: Double = fields(3).toDouble
            // 生成ROWRDD
            Row(id, name, age, fv)
        })
        //结果类型，其实就是表头，用于描述DataFrame
        val scme = StructType(List(
            StructField("id", LongType, true),
            StructField("name", StringType, true),
            StructField("age", IntegerType, true),
            StructField("fv", DoubleType, true)
        ))

        //将RowRDD关联schema
        val bdf: DataFrame = sqlContext.createDataFrame(rowRDD, scme)
        bdf.registerTempTable("t_boy")
        val result: DataFrame = sqlContext.sql("select * from t_boy order by fv desc, age asc")
        result.show()
        sc.stop()
    }
}

　　4.4 操作DataFrameAPI的形式进行数据操作

package cn.edu360.spark06

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

object spark1DataFrameOperate {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("spark1DataFrameOperate").setMaster("local[2]")
        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)
        val lines: RDD[String] = sc.textFile("hdfs://hd1:9000/sparkSQL/")
        val rowRDD: RDD[Row] = lines.map(line => {
            val fields: Array[String] = line.split(",")
            val id: Long = fields(0).toLong
            val name: String = fields(1)
            val age: Int = fields(2).toInt
            val fv: Double = fields(3).toDouble
            // 生成ROWRDD
            Row(id, name, age, fv)
        })
        //结果类型，其实就是表头，用于描述DataFrame
        val scme = StructType(List(
            StructField("id", LongType, true),
            StructField("name", StringType, true),
            StructField("age", IntegerType, true),
            StructField("fv", DoubleType, true)
        ))

        val bdf: DataFrame = sqlContext.createDataFrame(rowRDD, scme)

        //不使用SQL的方式，就不用注册临时表了
        val df1: DataFrame = bdf.select("name", "age", "fv")
        df1.show()
        
        // 操作数据的函数需要导入
        import sqlContext.implicits._
        val df2: Any = bdf.orderBy($"fv" desc, $"age" asc)
    }
}

　　4.5 Spark2.X的sql实现方式

package cn.edu360.spark06

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}


object spark2XDemo1 {
    def main(args: Array[String]): Unit = {
        //spark2.x SQL的编程API(SparkSession)
        //是spark2.x SQL执行的入口
        val sparkSession: SparkSession = SparkSession.builder()
            .appName("SQLTest1")
            .master("local[2]")
            .getOrCreate()
        //创建RDD
        val lines: RDD[String] = sparkSession.sparkContext.textFile("hdfs://hd1:9000/sparkSQL")
        val rowRDD: RDD[Row] = lines.map(line => {
            val fields = line.split(",")
            val id = fields(0).toLong
            val name = fields(1)
            val age = fields(2).toInt
            val fv = fields(3).toDouble
            Row(id, name, age, fv)
        })
        val scm = StructType(List(
            StructField("id", LongType, true),
            StructField("name", StringType, true),
            StructField("age", IntegerType, true),
            StructField("fv", DoubleType, true)
        ))
        val df: DataFrame = sparkSession.createDataFrame(rowRDD, scm)

        df.registerTempTable("t_boy")
        val result: DataFrame = sparkSession.sql("select * from t_boy")

        val result2: DataFrame = df.select("name", "age", "fv")

        import sparkSession.implicits._
        val result3: Dataset[Row] = df.orderBy($"fv" desc, $"age" asc)
        result.show()
        result2.show()
        result3.show()
        sparkSession.stop()
    }

}

　　4.6 SparkDataSet

　　Dataset分布式数据集，是对RDD的进一步封装，是更加智能的RDD, 在这进行wordcount实例：

package cn.edu360.spark06

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object sparkDateSet1 {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession.builder()
            .appName("sparkDateSet1")
            .master("local[2]")
            .getOrCreate()
        val lines: Dataset[String] = spark.read.textFile("hdfs://hd1:9000/wordcount/input/")

        import spark.implicits._
        val words: Dataset[String] = lines.flatMap(_.split(" "))

        // 通过操作Dataset进行数据操作
        val result: DataFrame = words.select("value")
        result.show()
        
        // 注册视图操作SQL形式
        words.createTempView("v_wc")
        val result2: DataFrame = spark.sql("select value, count(*) from v_wc group by value")
        result2.show()
        spark.stop()
    }
}