大数据之Spark（三）

Spark SQL基础

Spark SQL简介

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

• 容易整合（集成）
• 
• 统一的数据访问方式
• 
• 兼容Hive
• 
• 标准的数据连接
• 

基本概念：Datasets和DataFrames

• DataFrame
  DataFrame是组织成命名列的数据集。它在概念上等同于关系数据库中的表，但在底层具有更丰富的优化。DataFrames可以从各种来源构建，
  例如：

• 结构化数据文件

• hive中的表

• 外部数据库或现有RDDs
  DataFrame API支持的语言有Scala，Java，Python和R。
  
  从上图可以看出，DataFrame多了数据的结构信息，即schema。RDD是分布式的 Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化

• Datasets
  Dataset是数据的分布式集合。Dataset是在Spark 1.6中添加的一个新接口，是DataFrame之上更高一级的抽象。它提供了RDD的优点（强类型化，使用强大的lambda函数的能力）以及Spark SQL优化后的执行引擎的优点。一个Dataset 可以从JVM对象构造，然后使用函数转换（map， flatMap，filter等）去操作。 Dataset API 支持Scala和Java。 Python不支持Dataset API。

创建DataFrames

通过Case Class创建DataFrames

①　定义case class（相当于表的结构：Schema）

注意：由于mgr和comm列中包含null值，简单起见，将对应的case class类型定义为String

②　将HDFS上的数据读入RDD，并将RDD与case Class关联

③　将RDD转换成DataFrames

④　通过DataFrames查询数据

使用SparkSession

①　什么是SparkSession

Apache Spark 2.0引入了SparkSession，其为用户提供了一个统一的切入点来使用Spark的各项功能，并且允许用户通过它调用DataFrame和Dataset相关API来编写Spark程序。最重要的是，它减少了用户需要了解的一些概念，使得我们可以很容易地与Spark交互。
在2.0版本之前，与Spark交互之前必须先创建SparkConf和SparkContext。然而在Spark 2.0中，我们可以通过SparkSession来实现同样的功能，而不需要显式地创建SparkConf, SparkContext 以及 SQLContext，因为这些对象已经封装在SparkSession中。

②　创建StructType，来定义Schema结构信息

注意，需要：import org.apache.spark.sql.types._

③　读入数据并且切分数据

④　将RDD中的数据映射成Row

注意，需要：import org.apache.spark.sql.Row

⑤　创建DataFrames

val df = spark.createDataFrame(rowRDD,myschema)

再举一个例子，使用JSon文件来创建DataFame

①　源文件：$SPARK_HOME/examples/src/main/resources/people.json

②　val df = spark.read.json(“源文件”)

③　查看数据和Schema信息

DataFrame操作

DataFrame操作也称为无类型的Dataset操作

• 查询所有的员工姓名
• 
• 查询所有的员工姓名和薪水，并给薪水加100块钱
• 
• 查询工资大于2000的员工
• 
• 求每个部门的员工人数
• 
• ==完整的例子，请参考：
  http://spark.apache.org/docs/2.1.0/api/scala/index.html#org.apache.spark.sql.Dataset

在DataFrame中使用SQL语句

• 将DataFrame注册成表（视图）：df.createOrReplaceTempView(“emp”)
• 执行查询：spark.sql(“select * from emp”).show
• spark.sql(“select * from emp where deptno=10”).show
• spark.sql(“select deptno,sum(sal) from emp group by deptno”).show

Global Temporary View

上面使用的是一个在Session生命周期中的临时views。在Spark SQL中，如果你想拥有一个临时的view，并想在不同的Session中共享，而且在application的运行周期内可用，那么就需要创建一个全局的临时view。并记得使用的时候加上global_temp作为前缀来引用它，因为全局的临时view是绑定到系统保留的数据库global_temp上。

①　创建一个普通的view和一个全局的view

df.createOrReplaceTempView(“emp1”)
df.createGlobalTempView(“emp2”)

②　在当前会话中执行查询，均可查询出结果。

spark.sql(“select * from emp1”).show
spark.sql(“select * from global_temp.emp2”).show

③　开启一个新的会话，执行同样的查询

spark.newSession.sql(“select * from emp1”).show （运行出错）
spark.newSession.sql(“select * from global_temp.emp2”).show

创建Datasets

DataFrame的引入，可以让Spark更好的处理结构数据的计算，但其中一个主要的问题是：缺乏编译时类型安全。为了解决这个问题，Spark采用新的Dataset API (DataFrame API的类型扩展)。

Dataset是一个分布式的数据收集器。这是在Spark1.6之后新加的一个接口，兼顾了RDD的优点（强类型，可以使用功能强大的lambda）以及Spark SQL的执行器高效性的优点。所以可以把DataFrames看成是一种特殊的Datasets，即：Dataset(Row)

创建DataSet，方式一：使用序列

1、定义case class
case class MyData(a:Int,b:String)

2、生成序列，并创建DataSet
val ds = Seq(MyData(1,“Tom”),MyData(2,“Mary”)).toDS

3、查看结果

创建DataSet，方式二：使用JSON数据

1、定义case class
case class Person(name: String, gender: String)

2、通过JSON数据生成DataFrame
val df = spark.read.json(sc.parallelize("""{“gender”: “Male”, “name”: “Tom”}""" :: Nil))

3、将DataFrame转成DataSet
df.as[Person].show
df.as[Person].collect

创建DataSet，方式三：使用HDFS数据

1、读取HDFS数据，并创建DataSet
val linesDS = spark.read.text(“hdfs://hadoop111:9000/data/data.txt”).as[String]

2、对DataSet进行操作：分词后，查询长度大于3的单词
val words = linesDS.flatMap(.split(" ")).filter(.length > 3)
words.show
words.collect

3、执行WordCount程序
val result = linesDS.flatMap(.split(" ")).map((,1)).groupByKey(x => x._1).count
result.show
排序：result.orderBy($“value”).show

---

使用数据源

通用的Load/Save函数

什么是parquet文件？

Parquet是列式存储格式的一种文件类型，列式存储有以下的核心：

• 可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量
• 压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码（例如Run Length Encoding和Delta Encoding）进一步节约存储空间。
• 只读取需要的列，支持向量运算，能够获取更好的扫描性能
• Parquet格式是Spark SQL的默认数据源，可通过spark.sql.sources.default配置

通用的Load/Save函数

• 读取Parquet文件
  val usersDF = spark.read.load("/root/resources/users.parquet")
• 查询Schema和数据
  
• 查询用户的name和喜爱颜色，并保存

usersDF.select($"name",$"favorite_color").write.save("/root/result/parquet")

• 验证结果
• 

显式指定文件格式：加载json格式

• 直接加载：val usersDF = spark.read.load("/root/resources/people.json") 会出错

• val usersDF = spark.read.format(“json”).load("/root/resources/people.json")

存储模式（Save Modes）

可以采用SaveMode执行存储操作，SaveMode定义了对数据的处理模式。需要注意的是，这些保存模式不使用任何锁定，不是原子操作。此外，当使用Overwrite方式执行时，在输出新数据之前原数据就已经被删除。SaveMode详细介绍如下表：

Demo：

• usersDF.select($“name”).write.save("/root/result/parquet1")
  –> 出错：因为/root/result/parquet1已经存在
• usersDF.select($“name”).write.mode(“overwrite”).save("/root/result/parquet1")

将结果保存为表

• usersDF.select($“name”).write.saveAsTable(“table1”)
  也可以进行分区、分桶等操作：partitionBy、bucketBy

Parquet文件

Parquet是一个列格式而且用于多个数据处理系统中。Spark SQL提供支持对于Parquet文件的读写，也就是自动保存原始数据的schema。当写Parquet文件时，所有的列被自动转化为nullable，因为兼容性的缘故。

• 案例
• 读入json格式的数据，将其转换成parquet格式，并创建相应的表来使用SQL进行查询。
• 

Schema的合并

Parquet支持Schema evolution（Schema演变，即：合并）。用户可以先定义一个简单的Schema，然后逐渐的向Schema中增加列描述。通过这种方式，用户可以获取多个有不同Schema但相互兼容的Parquet文件。
Demo:

JSON Datasets

Spark SQL能自动解析JSON数据集的Schema，读取JSON数据集为DataFrame格式。读取JSON数据集方法为SQLContext.read().json()。该方法将String格式的RDD或JSON文件转换为DataFrame。
需要注意的是，这里的JSON文件不是常规的JSON格式。JSON文件每一行必须包含一个独立的、自满足有效的JSON对象。如果用多行描述一个JSON对象，会导致读取出错。读取JSON数据集示例如下：

• Demo1：使用Spark自带的示例文件 --> people.json 文件

定义路径：
val path ="/root/resources/people.json"

读取Json文件，生成DataFrame：
val peopleDF = spark.read.json(path)

打印Schema结构信息：
peopleDF.printSchema()

创建临时视图：
peopleDF.createOrReplaceTempView(“people”)

执行查询
spark.sql(“SELECT name FROM people WHERE age=19”).show

使用JDBC

Spark SQL同样支持通过JDBC读取其他数据库的数据作为数据源。
Demo演示：使用Spark SQL读取Oracle数据库中的表。

• 启动Spark Shell的时候，指定Oracle数据库的驱动

spark-shell --master spark://spark81:7077            \\
         --jars /root/temp/ojdbc6.jar              \\
         --driver-class-path /root/temp/ojdbc6.jar

读取Oracle数据库中的数据

• 方式一
  val oracleDF = spark.read.format(“jdbc”).
  option(“url”,“jdbc:oracle:thin:@192.168.88.101:1521/orcl.example.com”).
  option(“dbtable”,“scott.emp”).
  option(“user”,“scott”).
  option(“password”,“tiger”).
  load

• 方式二
  导入需要的类：
  import java.util.Properties

定义属性：
val oracleprops = new Properties()
oracleprops.setProperty(“user”,“scott”)
oracleprops.setProperty(“password”,“tiger”)

读取数据：
val oracleEmpDF =
spark.read.jdbc(“jdbc:oracle:thin:@192.168.88.101:1521/orcl.example.com”,
“scott.emp”,oracleprops)

使用Hive Table

• 首先，搭建好Hive的环境（需要Hadoop）

• 配置Spark SQL支持Hive

• 只需要将以下文件拷贝到$SPARK_HOME/conf的目录下，即可

• $HIVE_HOME/conf/hive-site.xml

• $HADOOP_CONF_DIR/core-site.xml

• $HADOOP_CONF_DIR/hdfs-site.xml

• 使用Spark Shell操作Hive

• 启动Spark Shell的时候，需要使用–jars指定mysql的驱动程序

• 创建表

• spark.sql(“create table src (key INT, value STRING) row format delimited fields terminated by ‘,’”)

• 导入数据

• spark.sql(“load data local path ‘/root/temp/data.txt’ into table src”)

• 查询数据

• spark.sql(“select * from src”).show

• 使用spark-sql操作Hive

• 启动spark-sql的时候，需要使用–jars指定mysql的驱动程序

• 操作Hive

• show tables;

• select * from 表;