四、spark--sparkSQL原理和使用

[TOC]

一、spark SQL概述

1.1 什么是spark SQL

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

1.2 spark SQL的特点

1、容易集成：安装Spark的时候，已经集成好了。不需要单独安装。
2、统一的数据访问方式：JDBC、JSON、Hive、parquet文件（一种列式存储文件，是SparkSQL默认的数据源，hive中也支持）
3、完全兼容Hive。可以将Hive中的数据，直接读取到Spark SQL中处理。
一般在生产中，基本都是使用hive做数据仓库存储数据，然后用spark从hive读取数据进行处理。
4、支持标准的数据连接：JDBC、ODBC
5、计算效率比基于mr的hive高，而且hive2.x版本中，hive建议使用spark作为执行引擎

二、spark SQL基本原理

2.1 DataFrame和DataSet基本概念

2.1.1 DataFrame

DataFrame是组织成命名列的数据集。它在概念上等同于关系数据库中的表，里面有表的结构以及数据，但在底层具有更丰富的优化。DataFrames可以从各种来源构建，
例如：
结构化数据文件
hive中的表
外部数据库或现有RDDs
DataFrame API支持的语言有Scala，Java，Python和R。

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

2.1.2 DataSet

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

2.2 创建DataFrame的方式

2.2.1 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中。
​ 要注意一点，在我用的这个spark版本中，直接使用new SQLContext() 来创建SQLContext对象，会显示该方式已经被弃用了（IDEA中会显示已弃用），建议使用SparkSession来获取SQLContext对象。

2.2.2 通过case class样本类

这种方式在scala中比较常用，因为case class是scala的特色

/**表 t_stu 的结构为：id name age*/object CreateDF {  def main(args: Array[String]): Unit = {    //这是最新的获取SQLContext对象的方式    //2、创建SparkSession对象，设置master，appnameval spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()    //3、通过spark获取sparkContext对象，读取数据    val lines = spark.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))    //4、将数据映射到case class中，也就是数据映射到表的对应字段中    val tb = lines.map(t=>emp(t(0).toInt,t(1),t(2).toInt))    //这里必须要加上隐式转换，否则无法调用 toDF 函数    import spark.sqlContext.implicits._    //5、生成df    val df2 = tb.toDF()    //相当于select name from t_stu    df1.select($"name").show()    //关闭spark对象    spark.stop()  }}/*1、定义case class，每个属性对应表中的字段名以及类型     一般生产中为了方便，会全部定义为string类型，然后有需要的时候     才根据实际情况将string转为需要的类型   这一步相当于定义表的结构*/case class emp(id:Int,name:String,age:Int)

总结步骤为：

1、定义case class，用来表结构2、创建sparkSession对象，用来读取数据3、将rdd中的数据和case class映射4、调用 toDF 函数将rdd转为 DataFrame

2.2.3 通过StructType类

这种方式java比较常用

package SparkSQLExerimport org.apacheimport org.apache.sparkimport org.apache.spark.sql.{Row, SparkSession}import org.apache.spark.{SparkConf, SparkContext}import org.apache.spark.sql.types.{DataTypes, StructField, StructType}/**  * 创建dataschema方式2：  * 通过spark session对象创建，表结构通过StructType创建  */object CreateDF02 {  def main(args: Array[String]): Unit = {    val sparkS = SparkSession.builder().master("local").appName("create schema").getOrCreate()    //1、通过StructType创建表结构schema，里面表的每个字段使用 StructField定义    val tbSchema = StructType(List(        StructField("id",DataTypes.IntegerType),        StructField("name",DataTypes.StringType),        StructField("age",DataTypes.IntegerType)      ))    //2、读取数据    var lines = sparkS.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))    //3、将数据映射为ROW对象    val rdd1 = lines.map(t=>Row(t(0).toInt,t(1),t(2).toInt))    //4、创建表结构和表数据映射，返回的就是df    val df2 = sparkS.createDataFrame(rdd1, tbSchema)    //打印表结构    df2.printSchema()    sparkS.stop()  }}

总结步骤为：

1、通过StructType创建表结构schema，里面表的每个字段使用 StructField定义2、通过sparkSession.sparkContext读取数据3、将数据映射格式为Row对象4、将StructType和数据Row对象映射，返回df

2.2.4 使用json等有表格式的文件类型

package SparkSQLExerimport org.apache.spark.sql.SparkSession/**  * 创建df方式3：通过有格式的文件直接导入数据以及表结构，比如json格式的文件  * 返回的直接就是一个DF  */object CreateDF03 {  def main(args: Array[String]): Unit = {    val sparkS = SparkSession.builder().master("local").appName("create df through json").getOrCreate()    //读取json方式1：    val jsonrdd1= sparkS.read.json("path")    //读取json方式2：    val jsonrdd1= sparkS.read.format("json").load("path")    sparkS.stop()  }}

这种方式比较简单，就是直接读取json文件而已
sparkS.read.xxxx读取任意文件时，返回的都是DF对象

2.3 操作DataFrame

2.3.1 DSL语句

DSL语句其实就是将sql语句的一些操作转为类似函数的方式去调用，比如：

df1.select("name").show

例子：

为了方便，直接在spark-shell里操作了，spark-shell --master spark://bigdata121:70771、打印表结构scala> df1.printSchemaroot|-- empno: integer (nullable = true)|-- ename: string (nullable = true)|-- job: string (nullable = true)|-- mgr: integer (nullable = true)|-- hiredate: string (nullable = true)|-- sal: integer (nullable = true)|-- comm: integer (nullable = true)|-- deptno: integer (nullable = true)2、显示当前df的表数据或者查询结果的数据scala> df1.show+-----+------+---------+----+----------+----+----+------+|empno| ename|      job| mgr|  hiredate| sal|comm|deptno|+-----+------+---------+----+----------+----+----+------+| 7369| SMITH|    CLERK|7902|1980/12/17| 800|   0|    20|| 7499| ALLEN| SALESMAN|7698| 1981/2/20|1600| 300|    30|| 7521|  WARD| SALESMAN|7698| 1981/2/22|1250| 500|    30|| 7566| JONES|  MANAGER|7839|  1981/4/2|2975|   0|    20|| 7654|MARTIN| SALESMAN|7698| 1981/9/28|1250|1400|    30|| 7698| BLAKE|  MANAGER|7839|  1981/5/1|2850|   0|    30|| 7782| CLARK|  MANAGER|7839|  1981/6/9|2450|   0|    10|| 7788| SCOTT|  ANALYST|7566| 1987/4/19|3000|   0|    20|| 7839|  KING|PRESIDENT|7839|1981/11/17|5000|   0|    10|| 7844|TURNER| SALESMAN|7698|  1981/9/8|1500|   0|    30|| 7876| ADAMS|    CLERK|7788| 1987/5/23|1100|   0|    20|| 7900| JAMES|    CLERK|7698| 1981/12/3| 950|   0|    30|| 7902|  FORD|  ANALYST|7566| 1981/12/3|3000|   0|    20|| 7934|MILLER|    CLERK|7782| 1982/1/23|1300|   0|    10|+-----+------+---------+----+----------+----+----+------+3、执行select, 相当于select xxx form  xxx where xxxscala> df1.select("ename","sal").where("sal>2000").show+------+----+| ename| sal|+------+----+| SMITH| 800|| ALLEN|1600||  WARD|1250|| JONES|2975||MARTIN|1250|| BLAKE|2850|| CLARK|2450|| SCOTT|3000||  KING|5000||TURNER|1500|| ADAMS|1100|| JAMES| 950||  FORD|3000||MILLER|1300|+------+----+4、对某些列进行操作对某个指定进行操作时，需要加上$符号，然后后面才能操作$代表 取出来以后，再做一些操作。注意：这个 $ 的用法在ideal中无法正常使用，解决方法下面说scala> df1.select($"ename",$"sal",$"sal"+100).show+------+----+-----------+| ename| sal|(sal + 100)|+------+----+-----------+| SMITH| 800|        900|| ALLEN|1600|       1700||  WARD|1250|       1350|| JONES|2975|       3075||MARTIN|1250|       1350|| BLAKE|2850|       2950|| CLARK|2450|       2550|| SCOTT|3000|       3100||  KING|5000|       5100||TURNER|1500|       1600|| ADAMS|1100|       1200|| JAMES| 950|       1050||  FORD|3000|       3100||MILLER|1300|       1400|+------+----+-----------+5、过滤行scala> df1.filter($"sal">2000).show+-----+-----+---------+----+----------+----+----+------+|empno|ename|      job| mgr|  hiredate| sal|comm|deptno|+-----+-----+---------+----+----------+----+----+------+| 7566|JONES|  MANAGER|7839|  1981/4/2|2975|   0|    20|| 7698|BLAKE|  MANAGER|7839|  1981/5/1|2850|   0|    30|| 7782|CLARK|  MANAGER|7839|  1981/6/9|2450|   0|    10|| 7788|SCOTT|  ANALYST|7566| 1987/4/19|3000|   0|    20|| 7839| KING|PRESIDENT|7839|1981/11/17|5000|   0|    10|| 7902| FORD|  ANALYST|7566| 1981/12/3|3000|   0|    20|+-----+-----+---------+----+----------+----+----+------+6、分组以及计数scala> df1.groupBy($"deptno").count.show+------+-----+                                                                  |deptno|count|+------+-----+|    20|    5||    10|    3||    30|    6|+------+-----+

上面说到在ide中 select($"name")中无法正常使用，解决方法为：

在该语句之前加上这么一句：import spark.sqlContext.implicits._主要还是因为类型的问题，加上隐式转换就好了

2.3.2 sql语句

df对象不能直接执行sql。需要生成一个视图，再执行SQL。
需要指定创建的视图名称，后面视图名称就相当于表名。
视图后面还会细说，这里先有个概念
例子：

val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()。。。。。。。。。。。。。。//通过df对象创建临时视图。视图名就相当于表名df1.createOrReplaceTempView("emp")//通过sparksession对象执行执行spark.sql("select * from emp").showspark.sql("select * from emp where sal > 2000").showspark.sql("select deptno,count(1) from emp group by deptno").show//可以创建多个视图，不冲突df1.createOrReplaceTempView("emp12345")spark.sql("select e.deptno from emp12345 e").show

2.3.3 多表查询

scala> case class Dept(deptno:Int,dname:String,loc:String)defined class Deptscala> val lines = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))lines: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[68] at map at :24scala> val allDept = lines.map(x=>Dept(x(0).toInt,x(1),x(2)))allDept: org.apache.spark.rdd.RDD[Dept] = MapPartitionsRDD[69] at map at :28scala> val df2 = allDept.toDFdf2: org.apache.spark.sql.DataFrame = [deptno: int, dname: string ... 1 more field]scala> df2.createcreateGlobalTempView   createOrReplaceTempView   createTempViewscala> df2.createOrReplaceTempView("dept")scala> spark.sql("select dname,ename from emp12345,dept where emp12345.deptno=dept.deptno").show+----------+------+                                                             |     dname| ename|+----------+------+|  RESEARCH| SMITH||  RESEARCH| JONES||  RESEARCH| SCOTT||  RESEARCH| ADAMS||  RESEARCH|  FORD||ACCOUNTING| CLARK||ACCOUNTING|  KING||ACCOUNTING|MILLER||     SALES| ALLEN||     SALES|  WARD||     SALES|MARTIN||     SALES| BLAKE||     SALES|TURNER||     SALES| JAMES|+----------+------+

2.4 创建DataSet

2.4.1 通过case class

和DataFrame类似，只是把 toDF改为调用 toDS 方法

package SparkSQLExerimport org.apache.spark.sql.SparkSessionimport org.apache.spark.{SparkConf, SparkContext}object CreateDS {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()    val lines = spark.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))    val tb = lines.map(t=>emp1(t(0).toInt,t(1),t(2).toInt))    import spark.sqlContext.implicits._    val df1 = tb.toDS()    df1.select($"name")  }}case class emp1(id:Int,name:String,age:Int)

2.4.2 通过序列Seq类对象

package SparkSQLExerimport org.apache.spark.sql.SparkSessionimport org.apache.spark.{SparkConf, SparkContext}object CreateDS {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()    //创建一个序列对象，里面都是emp1对象，映射的数据，然后直接toDS转为DataSet    val ds1 = Seq(emp1(1,"king",20)).toDS()    ds1.printSchema()  }}case class emp1(id:Int,name:String,age:Int)

2.4.3 使用json格式文件

定义case classcase class Person(name:String,age:BigInt)使用JSON数据生成DataFrameval df = spark.read.format("json").load("/usr/local/tmp_files/people.json")将DataFrame转换成DataSetdf.as[Person].showdf.as[Person] 是一个 DataSetas[T]中的泛型需要是一个case class类，用于映射表头

2.5 操作DataSet

DataSet支持的算子其实就是rdd和DataFrame算子的结合。

使用emp.json 生成DataFrameval empDF = spark.read.json("/usr/local/tmp_files/emp.json")scala> empDF.show+----+------+-----+------+----------+---------+----+----+|comm|deptno|empno| ename|  hiredate|      job| mgr| sal|+----+------+-----+------+----------+---------+----+----+|    |    20| 7369| SMITH|1980/12/17|    CLERK|7902| 800|| 300|    30| 7499| ALLEN| 1981/2/20| SALESMAN|7698|1600|| 500|    30| 7521|  WARD| 1981/2/22| SALESMAN|7698|1250||    |    20| 7566| JONES|  1981/4/2|  MANAGER|7839|2975||1400|    30| 7654|MARTIN| 1981/9/28| SALESMAN|7698|1250||    |    30| 7698| BLAKE|  1981/5/1|  MANAGER|7839|2850||    |    10| 7782| CLARK|  1981/6/9|  MANAGER|7839|2450||    |    20| 7788| SCOTT| 1987/4/19|  ANALYST|7566|3000||    |    10| 7839|  KING|1981/11/17|PRESIDENT|    |5000||   0|    30| 7844|TURNER|  1981/9/8| SALESMAN|7698|1500||    |    20| 7876| ADAMS| 1987/5/23|    CLERK|7788|1100||    |    30| 7900| JAMES| 1981/12/3|    CLERK|7698| 950||    |    20| 7902|  FORD| 1981/12/3|  ANALYST|7566|3000||    |    10| 7934|MILLER| 1982/1/23|    CLERK|7782|1300|+----+------+-----+------+----------+---------+----+----+scala> empDF.where($"sal" >= 3000).show+----+------+-----+-----+----------+---------+----+----+|comm|deptno|empno|ename|  hiredate|      job| mgr| sal|+----+------+-----+-----+----------+---------+----+----+|    |    20| 7788|SCOTT| 1987/4/19|  ANALYST|7566|3000||    |    10| 7839| KING|1981/11/17|PRESIDENT|    |5000||    |    20| 7902| FORD| 1981/12/3|  ANALYST|7566|3000|+----+------+-----+-----+----------+---------+----+----+#### empDF 转换成 DataSet 需要 case classscala> case class Emp(empno:BigInt,ename:String,job:String,mgr:String,hiredate:String,sal:BigInt,comm:String,deptno:BigInt)defined class Empscala> val empDS = empDF.as[Emp]empDS: org.apache.spark.sql.Dataset[Emp] = [comm: string, deptno: bigint ... 6 more fields]scala> empDS.filter(_.sal > 3000).show+----+------+-----+-----+----------+---------+---+----+|comm|deptno|empno|ename|  hiredate|      job|mgr| sal|+----+------+-----+-----+----------+---------+---+----+|    |    10| 7839| KING|1981/11/17|PRESIDENT|   |5000|+----+------+-----+-----+----------+---------+---+----+scala> empDS.filter(_.deptno == 10).show+----+------+-----+------+----------+---------+----+----+|comm|deptno|empno| ename|  hiredate|      job| mgr| sal|+----+------+-----+------+----------+---------+----+----+|    |    10| 7782| CLARK|  1981/6/9|  MANAGER|7839|2450||    |    10| 7839|  KING|1981/11/17|PRESIDENT|    |5000||    |    10| 7934|MILLER| 1982/1/23|    CLERK|7782|1300|+----+------+-----+------+----------+---------+----+----+

多表查询：

1、创建部门表scala> val deptRDD = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))deptRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[154] at map at :24scala> case class Dept(deptno:Int,dname:String,loc:String)defined class Deptscala> val deptDS = deptRDD.map(x=>Dept(x(0).toInt,x(1),x(2))).toDSdeptDS: org.apache.spark.sql.Dataset[Dept] = [deptno: int, dname: string ... 1 more field]scala> deptDS.show+------+----------+--------+|deptno|     dname|     loc|+------+----------+--------+|    10|ACCOUNTING|NEW YORK||    20|  RESEARCH|  DALLAS||    30|     SALES| CHICAGO||    40|OPERATIONS|  BOSTON|+------+----------+--------+2、员工表同上 empDSempDS.join(deptDS,"deptno").where(xxxx) 连接两个表，通过deptno字段empDS.joinWith(deptDS,deptDS("deptno")===empDS("deptno")) 这个用于连接的字段名称不一样的情况

2.6 视图view

​ 如果想使用标准的sql语句来操作df或者ds对象时，必须先给df或者ds对象创建视图，然后通过SparkSession对象的sql函数来对相应的视图进行操作才可以。那么视图是什么？
​ 视图是一个虚表，不存储数据，可以当做是表的一个访问链接。视图有两种类型：
普通视图：也叫本地视图，只在当前session会话中有效
全局视图：在全部session中都有效，全局视图创建在指定命名空间中：global_temp 类似于一个库
操作说明：

val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()val empDF = spark.read.json("/usr/local/tmp_files/emp.json")创建本地视图：empDF.createOrReplaceTempView(视图名)，视图存在就会重新创建empDF.createTempView(视图名)，如果视图存在就不会创建创建全局视图：empDF.createGlobalTempView(视图名)对视图执行sql操作,这里视图名就类似于表名spark.sql("xxxxx")例子：empDF.createOrReplaceTempView("emp")spark.sql("select * from emp").show注意，只要创建了视图，那么就可以通过sparksession对象在任意一个类中操作视图，也就是表。这个特性很好用，当我们要操作一些表时，可以一开始就读取成df，然后创建成视图，那么就可以在任意一个地方查询表了。

2.7 数据源

通过SparkSession对象可以读取不同格式的数据源：

val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()

下面都用上面的spark代称SparkSession。

2.7.1 SparkSession读取数据的方式

1、loadspark.read.load(path):读取指定路径的文件，要求文件存储格式为Parquet文件2、formatspark.read.format("格式").load(path) :指定读取其他格式的文件，如json例子：spark.read.format("json").load(path)3、直接读取其他格式文件spark.read.格式名(路径)，这是上面2中的一个简写方式，例子：spark.read.json(路径)  json格式文件spark.read.text(路径)  读取文本文件注意：这些方式返回的都是 DataFrame 对象

2.7.2 SparkSession保存数据的方式

可以将DataFrame 对象写入到指定格式的文件中，假设有个DataFrame 对象为df1.1、savedf1.write.save(路径) 他会将文件保存到这个目录下，文件名spark随机生成的，所以使用上面的读取方式的时候，直接指定读取目录即可，不用指定文件名。输出的文件格式为 Parquet。可以直接指定hdfs的路径，否则就存储到本地如：df1.write.save("/test")spark.read.load("/test")2、直接指定格式存储df1.write.json(路径)  这样就会以json格式保存文件，生成的文件名的情况和上面类似3、指定保存模式如果没有指定保存模式，输出路径存在的情况下，就会报错df1.write.mode("append").json(路径) mode("append") 就表示文件存在时就追加mode("overwrite") 表示覆盖旧数据4、保存为表df1.write.saveAsTable(表名) 会保存在当前目录的spark-warehouse 目录下5、formatdf1.write.format(格式).save()使用指定特定格式的方式来输出保存数据，比如保存到MongoDB数据库中

2.7.3 Parquet格式

​ 这种一种列式存储格式，具体原理可以看看之前hive的文章。这种格式是默认的存储格式，使用load和save时默认的格式，操作方式很像前面说的，这里不重复。这里要讲的是Parquet的一个特殊的功能，支持schema（表结构）的合并。例子：

scala> val df1 = sc.makeRDD(1 to 5).map(i=>(i,i*2)).toDF("single","double")df1: org.apache.spark.sql.DataFrame = [single: int, double: int]scala> df1.show+------+------+|single|double|+------+------+|     1|     2||     2|     4||     3|     6||     4|     8||     5|    10|+------+------+scala> sc.makeRDD(1 to 5)res8: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[26] at makeRDD at :25scala> sc.makeRDD(1 to 5).collectres9: Array[Int] = Array(1, 2, 3, 4, 5)//导出表1scala> df1.write.parquet("/usr/local/tmp_files/test_table/key=1")scala> val df2 = sc.makeRDD(6 to 10).map(i=>(i,i*3)).toDF("single","triple")df2: org.apache.spark.sql.DataFrame = [single: int, triple: int]scala> df2.show+------+------+|single|triple|+------+------+|     6|    18||     7|    21||     8|    24||     9|    27||    10|    30|+------+------+//导出表2scala> df2.write.parquet("/usr/local/tmp_files/test_table/key=2")scala> val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 1 more field]//直接读取会丢失字段scala> df3.show+------+------+---+|single|double|key|+------+------+---+|     8|  null|  2||     9|  null|  2||    10|  null|  2||     3|     6|  1||     4|     8|  1||     5|    10|  1||     6|  null|  2||     7|  null|  2||     1|     2|  1||     2|     4|  1|+------+------+---+//加上option，指定"mergeSchema"为true，就可以合并scala> val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 2 more fields]scala> df3.show+------+------+------+---+|single|double|triple|key|+------+------+------+---+|     8|  null|    24|  2||     9|  null|    27|  2||    10|  null|    30|  2||     3|     6|  null|  1||     4|     8|  null|  1||     5|    10|  null|  1||     6|  null|    18|  2||     7|  null|    21|  2||     1|     2|  null|  1||     2|     4|  null|  1|+------+------+------+---+补充问题：key 是什么？必须用key嘛？key是不同表的一个区分字段，在合并的时候，会作为合并之后的表的一个字段，并且值等于key=xx 中设置的值如果目录下，两个表的目录名不一样，是无法合并的，合并字段名可以任意，如：一个是key ，一个是 test 这两个无法合并，必须统一key或者test

2.7.4 json文件

这种一种带表格式字段的文件，例子：

scala> val peopleDF = spark.read.json("/usr/local/tmp_files/people.json")peopleDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> peopleDF.printSchema()root|-- age: long (nullable = true)|-- name: string (nullable = true)scala> peopleDF.createOrReplaceTempView("people")scala> spark.sql("select * from people where age=19")res25: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> spark.sql("select * from people where age=19").show+---+------+|age|  name|+---+------+| 19|Justin|+---+------+scala> spark.sql("select age,count(1) from people group by  age").show+----+--------+                                                                 | age|count(1)|+----+--------+|  19|       1||null|       1||  30|       1|+----+--------+

2.7.5 JDBC 连接

df对象支持通过jdbc连接数据库，写入数据到数据库，或者从数据库读取数据。
例子：
1、通过jdbc 从mysql读取数据：

使用 format(xx).option()的方式指定连接数据库的一些参数，比如用户名密码，使用的连接驱动等

import java.util.Propertiesimport org.apache.spark.sql.SparkSessionobject ConnMysql {  def main(args: Array[String]): Unit = {    val sparkS = SparkSession.builder().appName("spark sql conn mysql").master("local").getOrCreate()    //连接mysql方式1：    //创建properties配置对象，用于存放连接mysql的参数    val mysqlConn = new Properties()    mysqlConn.setProperty("user","root")    mysqlConn.setProperty("password","wjt86912572")    //使用jdbc连接,指定连接字符串，表名，以及其他连接参数，并返回对应的dataframeval mysqlDF1 = sparkS.read.jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8","customer",mysqlConn)    mysqlDF1.printSchema()    mysqlDF1.show()    mysqlDF1.createTempView("customer")    sparkS.sql("select * from customer limit 2").show()    //连接mysql方式2，这种方式比较常用：    val mysqlConn2 = sparkS.read.format("jdbc")      .option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8")                    .option("user","root")      .option("password","wjt86912572")      .option("driver","com.mysql.jdbc.Driver")      .option("dbtable","customer").load()    mysqlConn2.printSchema()  }}

这是两种连接读取数据的方式。

2、jdbc写入数据到mysql

和读取类似，只不过换成了write操作

import java.util.Propertiesimport org.apache.spark.sql.SparkSessionobject WriteToMysql {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder().appName("write to mysql").master("local").getOrCreate()    val df1 = spark.read.text("G:\\test\\t_stu.json")    //方式1：    df1.write.format("jdbc")      .option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8")      .option("user","root")      .option("password","wjt86912572")      .option("driver","com.mysql.jdbc.Driver")      .option("dbtable","customer").save()    //方式2：    val mysqlConn = new Properties()    mysqlConn.setProperty("user","root")    mysqlConn.setProperty("password","wjt86912572")    df1.write.jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8","customer",mysqlConn)  }}必须要保证df的表格式和写入的mysql的表格式一样，字段名也要一样

2.7.6 hive

1、通过jdbc连接hive
方式和普通jdbc类似，例如：

import java.util.Propertiesimport org.apache.spark.sql.SparkSession/**  * 连接hive的情况有两种：  * 1、如果是直接在ideal中运行spark程序的话，则必须在程序中指定jdbc连接的hiveserver的地址  * 且hiveserver必须以后台服务的形式暴露10000端口出来.这种方式是直接通过jdbc连接hive  *  * 2、如果程序是打包到spark集群中运行的话，一般spark集群的conf目录下，已经有hive client  * 的配置文件，就会直接启动hive client来连接hive。这时不需要启动hiveserver服务。  * 这种方式是通过hive client连接hive  */object ConnHive {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder().appName("spark sql conn mysql").master("local").getOrCreate()    val properties = new Properties()    properties.setProperty("user","")    properties.setProperty("password","")    val hiveDF = spark.read.jdbc("jdbc:hive2://bigdata121:10000/default","customer",properties)    hiveDF.printSchema()    spark.stop()  }}

这种方式要注意一点：
hiveserver必须以后台服务的形式暴露10000端口出来.这种方式是直接通过jdbc连接hive。

2、通过hive client连接hive
​ 这种方式一般用在生产中，因为任务一般都是通过spark-submit提交到集群中运行，这时候就会直接通过hive client来连接hive，不会通过jdbc来连接了。
​ 要注意：需要在spark的节点上都配置上hive client，然后将hive-site.xml配置文件拷贝到 spark的conf目录下。同时需要将hadoop的core-site.xml hdfs-site.xml也拷贝过去。另外一方面，因为要使用hive client，所以hive server那边，一般都要配置metastore server，具体配置看hive的文章。
​ 这样在spark集群中的程序就可以直接使用

spark.sql("xxxx").show这样的操作，默认就会从hive中读取对应的表进行操作。不用另外做任何连接hive 的操作

或者直接到 spark-shell中，也是可以直接使用 上面的方式操作hive的表
例如：

import org.apache.spark.sql.SparkSessionobject ConnHive02 {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder().appName("spark sql conn hive").getOrCreate()    spark.sql("select * from customer").show()  }}这样直接操作的就是 hive 的表了

2.8 小案例--读取hive数据分析结果写入mysql

import java.util.Propertiesimport org.apache.spark.sql.SparkSessionobject HiveToMysql {  def main(args: Array[String]): Unit = {    //直接通过spark集群中的hive client连接hive，不需要jdbc以及hive server    val spark = SparkSession.builder().appName("hive to mysql").enableHiveSupport().getOrCreate()    val resultDF = spark.sql("select * from default.customer")    //一般中间写的处理逻辑都是处理从hive读取的数据，处理完成后写入到mysql    val mysqlConn = new Properties()    mysqlConn.setProperty("user","root")    mysqlConn.setProperty("password","wjt86912572")    //通过jdbc写入mysql  resultDF.write.mode("append").jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8", "customer", mysqlConn)    spark.stop()  }}

三、性能优化

3.1 内存中缓存数据

先启动个spark-shell

spark-shell --master spark://bigdata121:7077要在spark-shell中操作mysql，所以记得自己找个 mysql-connector的jar放到spark的jars目录下

例子：

创建df，从mysql读取表scala> val mysqDF = spark.read.format("jdbc").option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8").option("user","root").option("password","wjt86912572").option("driver","com.mysql.jdbc.Driver").option("dbtable","customer").load()mysqDF: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> mysqDF.show+---+------+--------------------+| id|  name|            last_mod|+---+------+--------------------+|  1|  neil|2019-07-20 17:09:...||  2|  jack|2019-07-20 17:09:...||  3|martin|2019-07-20 17:09:...||  4|  tony|2019-07-20 17:09:...||  5|  eric|2019-07-20 17:09:...||  6|  king|2019-07-20 17:42:...||  7|   tao|2019-07-20 17:45:...|+---+------+--------------------+必须注册成一张表，才可以缓存。scala> mysqDF.registerTempTable("customer")warning: there was one deprecation warning; re-run with -deprecation for details标识这张表可以被缓存，但是现在数据并没有直接缓存scala> spark.sqlContext.cacheTable("customer")第一次查询表，从mysql读取数据，并缓存到内存中scala> spark.sql("select * from customer").show+---+------+--------------------+| id|  name|            last_mod|+---+------+--------------------+|  1|  neil|2019-07-20 17:09:...||  2|  jack|2019-07-20 17:09:...||  3|martin|2019-07-20 17:09:...||  4|  tony|2019-07-20 17:09:...||  5|  eric|2019-07-20 17:09:...||  6|  king|2019-07-20 17:42:...||  7|   tao|2019-07-20 17:45:...|+---+------+--------------------+这一次查询从内存中返回scala> spark.sql("select * from customer").show+---+------+--------------------+| id|  name|            last_mod|+---+------+--------------------+|  1|  neil|2019-07-20 17:09:...||  2|  jack|2019-07-20 17:09:...||  3|martin|2019-07-20 17:09:...||  4|  tony|2019-07-20 17:09:...||  5|  eric|2019-07-20 17:09:...||  6|  king|2019-07-20 17:42:...||  7|   tao|2019-07-20 17:45:...|+---+------+--------------------+清空缓存scala> spark.sqlContext.clearCache

3.2 调优相关参数

将数据缓存到内存中的相关优化参数   spark.sql.inMemoryColumnarStorage.compressed   默认为 true   Spark SQL 将会基于统计信息自动地为每一列选择一种压缩编码方式。   spark.sql.inMemoryColumnarStorage.batchSize   默认值：10000   缓存批处理大小。缓存数据时, 较大的批处理大小可以提高内存利用率和压缩率，但同时也会带来 OOM（Out Of Memory）的风险。其他性能相关的配置选项（不过不推荐手动修改，可能在后续版本自动的自适应修改）   spark.sql.files.maxPartitionBytes   默认值：128 MB   读取文件时单个分区可容纳的最大字节数   spark.sql.files.openCostInBytes   默认值：4M   打开文件的估算成本, 按照同一时间能够扫描的字节数来测量。当往一个分区写入多个文件的时候会使用。高估更好, 这样的话小文件分区将比大文件分区更快 (先被调度)。spark.sql.autoBroadcastJoinThreshold   默认值：10M   用于配置一个表在执行 join 操作时能够广播给所有 worker 节点的最大字节大小。通过将这个值设置为 -1 可以禁用广播。注意，当前数据统计仅支持已经运行了 ANALYZE TABLE  COMPUTE STATISTICS noscan 命令的 Hive Metastore 表。spark.sql.shuffle.partitions   默认值：200   用于配置 join 或聚合操作混洗（shuffle）数据时使用的分区数。