Spark SQL（Spark shell）

2020-10-14 本文已影响0人羋学僧

一、Spark SQL基础

1、Spark SQL简介

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

为什么要学习Spark SQL？

Hive，它是将Hive SQL转换成MapReduce然后提交到集群上执行，大大简化了编写MapReduce的程序的复杂性，由于MapReduce这种计算模型执行效率比较慢。所以Spark SQL的应运而生，它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快！同时Spark SQL也支持从Hive中读取数据。

Spark SQL的特点：

容易整合（集成）

统一的数据访问方式

兼容Hive

标准的数据连接

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， ﬂatMap，ﬁlter 等）去操作。 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

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

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

val lines = sc.textFile("hdfs://bigdata02:9000/emp.csv").map(_.split(","))

val allEmp =lines.map(x=>Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt))

③ 将RDD转换成DataFrames

val empDF =allEmp.toDF

④ 通过DataFrames查询数据

SELECT

empDF.show

DESCRIBE

empDF.printSchema

使用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._
import org.apache.spark.sql.types._

val myschema = StructType(List(StructField("empno", DataTypes.IntegerType), StructField("ename", DataTypes.StringType),StructField("job", DataTypes.StringType),StructField("mgr", DataTypes.StringType),StructField("hiredate", DataTypes.StringType),StructField("sal", DataTypes.IntegerType),StructField("comm", DataTypes.StringType),StructField("deptno", DataTypes.IntegerType)))

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

③ 读入数据并且切分数据

val lines = sc.textFile("hdfs://bigdata02:9000/emp.csv").map(_.split(","))

④ 将RDD中的数据映射成Row

val rowRDD =lines.map(x=>Row(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt))

注意，需要：import org.apache.spark.sql.Row,前面已经导入

⑤ 创建DataFrames

val df = spark.createDataFrame(rowRDD,myschema)

df.show

使用JSon文件来创建DataFame

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

cat /home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json

② 读取数据

val peopleDF = spark.read.json("/home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json")

③ 查看数据和Schema信息

peopleDF.show

5、DataFrame操作

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

查询所有的员工姓名

empDF.select("ename").show 
empDF.select($"ename").show

查询所有的员工姓名和薪水，并给薪水加100块钱

empDF.select($"ename",$"sal",$"sal"+100).show

查询工资大于2000的员工

empDF.filter($"sal">2000).show

求每个部门的员工人数

empDF.groupBy($"deptno").count.show

完整的例子，请参考：Dataset

在DataFrame中使用SQL语句
使用SQL前提条件:我们需要把DataFrame注册成是一个table或者view
① 将DataFrame注册成表（视图）：

empDF.createOrReplaceTempView("emp")

② 执行查询：

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

查询10号部门员工

spark.sql("select * from emp where deptno =10").show

分组
求部门工资总和
部门号、工资总和

spark.sql("select deptno,sum(sal) from emp group by deptno").show

二、使用数据源

1、使用load和save函数

什么是parquet文件？

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

load加载数据

a.默认文件格式 parquet

val userDF=spark.read.load("/home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/users.parquet")

val peopleDF =spark.read.json("/home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json")

val peopleDF =spark.read.format("json").load("/home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json")

save保存数据(默认写出去parquet文件)

查询一下用户的名字和喜欢的颜色并保存

userDF.select($"name",$"favorite_color").write.save("/home/bigdata/data/result1015")

userDF.select($"name",$"favorite_color").write.format("csv").save("/home/bigdata/data/result0927")

2、使用parquet数据文件

a.格式转换

有没有方法把别的文件转换成parquet文件?

val empjson = spark.read.json("/home/bigdata/data/emp.json")
empjson.write.parquet("/home/bigdata/data/parquet0927")

b.schema合并

RDD-->DataFrame-->parquet

val df1=sc.makeRDD(1 to 5).map(i=>(i,i*2)).toDF("single","double")

df1.write.parquet("/home/bigdata/data/test_table/key=1")

val df2=sc.makeRDD(6 to 10).map(i=>(i,i*3)).toDF("single","triple")

df2.write.parquet("/home/bigdata/data/test_table/key=2")

val df3 = spark.read.option("mergeSchema","true").parquet("/home/bigdata/data/test_table")

df2.write.parquet("hdfs://bigdata03:9000/test_table/key=2")

df1.write.parquet("hdfs://bigdata03:9000/test_table/key=1")

val df3 = spark.read.option("mergeSchema","true").parquet("hdfs://bigdata03:9000/test_table")

3、使用json数据文件

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 ="/home/bigdata/apps/spark-2.1.0-bin-hadoop2.7/examples/src/main/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读取MySQL数据库中的表。

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

bin/spark-shell --master spark://bigdata02:7077 --jars /home/bigdata/data/mysql-connector-java-5.1.40-bin.jar --driver-class-path /home/bigdata/data/mysql-connector-java-5.1.40-bin.jar

读取MySQL数据库中的数据

方式一：

val mysqlDF = spark.read.format("jdbc").option("url","jdbc:mysql://bigdata02:3306/sqoopdb").option("dbtable","sqoopstudent").option("user","bigdata").option("password","123456").load

方式二：

导入需要的类：

import java.util.Properties

定义属性：

val mysqlprops = new Properties()

mysqlprops.setProperty("user","bigdata")

mysqlprops.setProperty("password","123456")

读取数据：

val mysql2DF = spark.read.jdbc("jdbc:mysql://bigdata02:3306/sqoopdb","sqoopstudent",mysqlprops)

5、使用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 student(id int, name string, sex string, age int, department string) row format delimited fields terminated by ','")

- 导入数据

spark.sql("load data inpath 'hdfs://bigdata02:9000/student.txt' into table student")  #报错

错误信息

java.lang.IllegalArgumentException: Wrong FS: hdfs://bigdata02:9000/student.txt, expected: hdfs://hadoop277ha
  at org.apache.hadoop.fs.FileSystem.checkPath(FileSystem.java:649)
  at org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:194)
  at org.apache.hadoop.hdfs.DistributedFileSystem.getEZForPath(DistributedFileSystem.java:2093)
  at org.apache.hadoop.hdfs.client.HdfsAdmin.getEncryptionZoneForPath(HdfsAdmin.java:289)
  at org.apache.hadoop.hive.shims.Hadoop23Shims$HdfsEncryptionShim.isPathEncrypted(Hadoop23Shims.java:1221)
  at org.apache.hadoop.hive.ql.metadata.Hive.moveFile(Hive.java:2607)
  at org.apache.hadoop.hive.ql.metadata.Hive.copyFiles(Hive.java:2711)
  at org.apache.hadoop.hive.ql.metadata.Hive.loadTable(Hive.java:1645)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:497)
  at org.apache.spark.sql.hive.client.Shim_v0_14.loadTable(HiveShim.scala:716)
  at org.apache.spark.sql.hive.client.HiveClientImpl$$anonfun$loadTable$1.apply$mcV$sp(HiveClientImpl.scala:672)
  at org.apache.spark.sql.hive.client.HiveClientImpl$$anonfun$loadTable$1.apply(HiveClientImpl.scala:672)
  at org.apache.spark.sql.hive.client.HiveClientImpl$$anonfun$loadTable$1.apply(HiveClientImpl.scala:672)
  at org.apache.spark.sql.hive.client.HiveClientImpl$$anonfun$withHiveState$1.apply(HiveClientImpl.scala:283)
  at org.apache.spark.sql.hive.client.HiveClientImpl.liftedTree1$1(HiveClientImpl.scala:230)
  at org.apache.spark.sql.hive.client.HiveClientImpl.retryLocked(HiveClientImpl.scala:229)
  at org.apache.spark.sql.hive.client.HiveClientImpl.withHiveState(HiveClientImpl.scala:272)
  at org.apache.spark.sql.hive.client.HiveClientImpl.loadTable(HiveClientImpl.scala:671)
  at org.apache.spark.sql.hive.HiveExternalCatalog$$anonfun$loadTable$1.apply$mcV$sp(HiveExternalCatalog.scala:741)
  at org.apache.spark.sql.hive.HiveExternalCatalog$$anonfun$loadTable$1.apply(HiveExternalCatalog.scala:739)
  at org.apache.spark.sql.hive.HiveExternalCatalog$$anonfun$loadTable$1.apply(HiveExternalCatalog.scala:739)
  at org.apache.spark.sql.hive.HiveExternalCatalog.withClient(HiveExternalCatalog.scala:95)
  at org.apache.spark.sql.hive.HiveExternalCatalog.loadTable(HiveExternalCatalog.scala:739)
  at org.apache.spark.sql.catalyst.catalog.SessionCatalog.loadTable(SessionCatalog.scala:319)
  at org.apache.spark.sql.execution.command.LoadDataCommand.run(tables.scala:302)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:58)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:56)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:74)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:135)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:132)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:113)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:87)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:87)
  at org.apache.spark.sql.Dataset.<init>(Dataset.scala:185)
  at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:64)
  at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:592)
  ... 48 elided

修改代码

spark.sql("load data inpath 'hdfs://hadoop277ha/student.txt' into table student")

- 查询数据

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

使用spark-sql操作Hive

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

- 操作Hive

 spark.sql("show tables").show
 spark.sql("select * from student").show