Spark学习之Spark SQL

Spark SQL

1、Spark SQL基础

1、Spark SQL简介

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

为何要学习Spark SQL？咱们已经学习了Hive，它是将Hive SQL转换成MapReduce而后提交到集群上执行，大大简化了编写MapReduce的程序的复杂性，因为MapReduce这种计算模型执行效率比较慢。因此Spark SQL的应运而生，它是将Spark SQL转换成RDD，而后提交到集群执行，执行效率很是快！同时Spark SQL也支持从Hive中读取数据。

 

Spark SQL的特色：

1.容易整合（集成） 

2.统一的数据访问方式

3.兼容Hive
 

4.标准的数据链接

2、基本概念：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。

3、测试数据

使用员工表的数据，并已经将其保存到了HDFS上。
emp.csv

dept.csv

4、建立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信息
 

5、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

 

6、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

 

7、建立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

 

8、Datasets的操做案例
　　emp.json

（*）使用emp.json 生成DataFrame

val empDF = spark.read.json("/root/resources/emp.json")

            查询工资大于3000的员工

            empDF.where($"sal" >= 3000).show

 

（*）建立case class

case class Emp(empno:Long,ename:String,job:String,hiredate:String,mgr:String,sal:Long,comm:String,deptno:Long)

 

（*）生成DataSets，并查询数据

     val empDS = empDF.as[Emp]

 

     查询工资大于3000的员工

     empDS.filter(_.sal > 3000).show

 

     查看10号部门的员工

     empDS.filter(_.deptno == 10).show

 

（*）多表查询

1、建立部门表

val deptRDD=sc.textFile("/root/temp/dept.csv").map(_.split(","))

case class Dept(deptno:Int,dname:String,loc:String)

val deptDS = deptRDD.map(x=>Dept(x(0).toInt,x(1),x(2))).toDS

 

2、建立员工表

case class Emp(empno:Int,ename:String,job:String,mgr:String,hiredate:String,sal:Int,comm:String,deptno:Int)

val empRDD = sc.textFile("/root/temp/emp.csv").map(_.split(","))

val empDS = empRDD.map(x => Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt)).toDS

 

3、执行多表查询：等值连接

    val result = deptDS.join(empDS,"deptno")

    

    另外一种写法：注意有三个等号

    val result = deptDS.joinWith(empDS,deptDS("deptno")=== empDS("deptno"))

    joinWith和join的区别是链接后的新Dataset的schema会不同

 

（*）查看执行计划：result.explain

 

2、使用数据源 

1、通用的Load/Save函数

（*）什么是parquet文件？

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

l 能够跳过不符合条件的数据，只读取须要的数据，下降IO数据量。

l 压缩编码能够下降磁盘存储空间。因为同一列的数据类型是同样的，可使用更高效的压缩编码（例如Run Length Encoding和Delta Encoding）进一步节约存储空间。

l 只读取须要的列，支持向量运算，可以获取更好的扫描性能。

l Parquet格式是Spark SQL的默认数据源，可经过spark.sql.sources.default配置

 

（*）通用的Load/Save函数

l 读取Parquet文件

val usersDF = spark.read.load("/root/resources/users.parquet")

l 查询Schema和数据

l 查询用户的name和喜好颜色，并保存

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

 

l 验证结果

（*）显式指定文件格式：加载json格式

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

                      会出错

l 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

2、Parquet文件

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

（*）案例：

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

（*）Schema的合并：

Parquet支持Schema evolution（Schema演变，即：合并）。用户能够先定义一个简单的Schema，而后逐渐的向Schema中增长列描述。经过这种方式，用户能够获取多个有不一样Schema但相互兼容的Parquet文件。

Demo:
 

3、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

 

4、使用JDBC

Spark SQL一样支持经过JDBC读取其余数据库的数据做为数据源。

Demo演示：使用Spark SQL读取Oracle数据库中的表。

l 启动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)

 

注意：下面是读取Oracle 10g（Windows 上）的步骤

5、使用Hive Table

l 首先，搭建好Hive的环境（须要Hadoop）

l 配置Spark SQL支持Hive

　　只须要将如下文件拷贝到$SPARK_HOME/conf的目录下，便可

　　$HIVE_HOME/conf/hive-site.xml

　　$HADOOP_CONF_DIR/core-site.xml

　　$HADOOP_CONF_DIR/hdfs-site.xml

 

l 使用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

 

l 使用spark-sql操做Hive

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

　　操做Hive

　　show tables;

　　select * from 表;

 

 

3、性能优化

1、在内存中缓存数据

性能调优主要是将数据放入内存中操做。经过spark.cacheTable("tableName")或者dataFrame.cache()。使用spark.uncacheTable("tableName")来从内存中去除table。

Demo案例：

（*）从Oracle数据库中读取数据，生成DataFrame

     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

 

（*）将DataFrame注册成表：    oracleDF.registerTempTable("emp")

 

（*）执行查询，并经过Web Console监控执行的时间

                 spark.sql("select * from emp").show
 

（*）将表进行缓存，并查询两次，并经过Web Console监控执行的时间

     spark.sqlContext.cacheTable("emp")
 

（*）清空缓存：

      spark.sqlContext.cacheTable("emp")

      spark.sqlContext.clearCache

2、性能优化相关参数

1.将数据缓存到内存中的相关优化参数

　　(1)spark.sql.inMemoryColumnarStorage.compressed

　　默认为 true

　　Spark SQL 将会基于统计信息自动地为每一列选择一种压缩编码方式。

 

　　(2)spark.sql.inMemoryColumnarStorage.batchSize

　　默认值：10000

　　缓存批处理大小。缓存数据时, 较大的批处理大小能够提升内存利用率和压缩率，但同时也会带来 OOM（Out Of Memory）的风险。

 

2.其余性能相关的配置选项（不过不推荐手动修改，可能在后续版本自动的自适应修改）

　　(1)spark.sql.files.maxPartitionBytes

　　默认值：128 MB

　　读取文件时单个分区可容纳的最大字节数

 

　　(2)spark.sql.files.openCostInBytes

　　默认值：4M

　　打开文件的估算成本, 按照同一时间可以扫描的字节数来测量。当往一个分区写入多个文件的时候会使用。高估更好, 这样的话小文件分区将比大文件分区更快 (先被调度)。

 

　　(3)spark.sql.autoBroadcastJoinThreshold

　　默认值：10M

　　用于配置一个表在执行 join 操做时可以广播给全部 worker 节点的最大字节大小。经过将这个值设置为 -1 能够禁用广播。注意，当前数据统计仅支持已经运行了 ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan 命令的 Hive Metastore 表。

 

　　(4)spark.sql.shuffle.partitions

　　默认值：200

　　用于配置 join 或聚合操做混洗（shuffle）数据时使用的分区数。

4、在IDEA中开发Spark SQL程序

 

1、指定Schema格式

package sparksql

import org.apache.spark.sql.{Row, SQLContext, SaveMode, SparkSession}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkConf, SparkContext}

object SpecifyingSchema {
  def main(args: Array[String]) {
      //建立Spark Session对象
    val spark = SparkSession.builder().master("local").appName("UnderstandingSparkSession").getOrCreate()

    //从指定的地址建立RDD
    val personRDD = spark.sparkContext.textFile("D:\\temp\\student.txt").map(_.split(" "))

    //经过StructType直接指定每一个字段的schema
    val schema = StructType(
      List(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
    //将RDD映射到rowRDD
    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))

    //将schema信息应用到rowRDD上
    val personDataFrame = spark.createDataFrame(rowRDD, schema)

    //注册表
    personDataFrame.createOrReplaceTempView("t_person")

    //执行SQL
    val df = spark.sql("select * from t_person order by age desc limit 4")

    //显示结果
    df.show()

    //中止Spark Context
    spark.stop()
  }
}

 

2、使用case class 

package demo

import org.apache.spark.sql.SparkSession

//使用case class
object Demo2 {

  def main(args: Array[String]): Unit = {
    //建立SparkSession
    val spark = SparkSession.builder().master("local").appName("My Demo 1").getOrCreate()

    //从指定的文件中读取数据，生成对应的RDD
    val lineRDD = spark.sparkContext.textFile("d:\\temp\\student.txt").map(_.split(" "))

    //将RDD和case class 关联
    val studentRDD = lineRDD.map( x => Student(x(0).toInt,x(1),x(2).toInt))

    //生成 DataFrame，经过RDD 生成DF,导入隐式转换
    import spark.sqlContext.implicits._
    val studentDF = studentRDD.toDF

    //注册表 视图
    studentDF.createOrReplaceTempView("student")

    //执行SQL
    spark.sql("select * from student").show()

    spark.stop()
  }
}

//case class 必定放在外面
case class Student(stuID:Int,stuName:String,stuAge:Int)

 

 

3、就数据保存到数据库

package demo

import java.util.Properties

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}


//做用：读取本地一个文件, 生成对应 DataFrame，注册表
object Demo1 {

  def main(args: Array[String]): Unit = {
    //建立SparkSession
    val spark = SparkSession.builder().master("local").appName("My Demo 1").getOrCreate()

    //从指定的文件中读取数据，生成对应的RDD
    val personRDD = spark.sparkContext.textFile("d:\\temp\\student.txt").map(_.split(" "))

    //建立schema ，经过StructType
    val schema = StructType(
        List(
          StructField("personID",IntegerType,true),
          StructField("personName",StringType,true),
          StructField("personAge",IntegerType,true)
        )
    )

    //将RDD映射到Row RDD 行的数据上
    val rowRDD = personRDD.map(p => Row(p(0).toInt,p(1).trim,p(2).toInt))

    //生成DataFrame
    val personDF = spark.createDataFrame(rowRDD,schema)

    //将DF注册成表
    personDF.createOrReplaceTempView("myperson")

    //执行SQL
    val result = spark.sql("select * from myperson")

    //显示
    //result.show()

    //将结果保存到oracle中
    val props = new Properties()
    props.setProperty("user","scott")
    props.setProperty("password","tiger")

    result.write.jdbc("jdbc:oracle:thin:@192.168.88.101:1521/orcl.example.com","scott.myperson",props)

    //若是表已经存在,append的方式数据
    //result.write.mode("append").jdbc("jdbc:oracle:thin:@192.168.88.101:1521/orcl.example.com","scott.myperson",props)


    //中止spark context
    spark.stop()
  }
}