《通信数据分析与实战》课件-第四章 Spark SQL

通信数据分析与实战SparkSQL第四章第1节2知道SparkSQL的作用掌握SparkSQL的架构学习目标TARGETSparkSQL的概述在很多情况下，开发工程师并不了解Scala语言，也不了解Spark常用API，但又非常想要使用Spark框架提供的强大的数据分析能力。Spark的开发工程师们考虑到了这个问题，利用SQL语言的语法简洁、学习门槛低以及在编程语言普及程度和流行程度高等诸多优势，从而开发了SparkSQL模块，通过SparkSQL，开发人员能够通过使用SQL语句，实现对结构化数据的处理。SparkSQL的概述SparkSQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象结构叫做DataFrame的数据模型（即带有Schema信息的RDD），SparkSQL作为分布式SQL查询引擎，让用户可以通过SQL、DataFramesAPI和DatasetsAPI三种方式实现对结构化数据的处理。SparkSQL的概述SparkSQL主要提供了以下三个功能：SparkSQL可从各种结构化数据源中读取数据，进行数据分析。SparkSQL包含行业标准的JDBC和ODBC连接方式，因此它不局限于在Spark程序内使用SQL语句进行查询。SparkSQL可以无缝地将SQL查询与Spark程序进行结合，它能够将结构化数据作为Spark中的分布式数据集（RDD）进行查询。SparkSQL的架构SparkSQL架构与Hive架构相比，把底层的MapReduce执行引擎更改为Spark，还修改了Catalyst优化器，SparkSQL快速的计算效率得益于Catalyst优化器。从HiveQL被解析成语法抽象树起，执行计划生成和优化的工作全部交给SparkSQL的Catalyst优化器进行负责和管理。SparkSQL架构图SparkSQL的架构Spark要想很好地支持SQL，需要完成解析（Parser）、优化（Optimizer）、执行（Execution）三大过程。SparkSQL的架构SqlParse：完成SQL语法解析功能，目前只提供了一个简单的SQL解析器。Analyze：主要完成绑定工作，将不同来源的UnresolvedLogicalPlan和元数据进行绑定，生成ResolvedLogicalPlan。Optimizer：对ResolvedLo;gicalPlan进行优化，生成OptimizedLogicalPlan。Planner：将LogicalPlan转换成PhysicalPlan。CostModel：主要根据过去的性能统计数据，选择最佳的物理执行计划。Catalyst优化器在执行计划生成和优化的工作时，离不开内部的五大组件。SparkSQL的架构在了解了上述组件的作用后，下面分步骤讲解SparkSQL工作流程。

SparkSQL的架构SparkSQL工作流程：1下在解析SQL语句之前，会创建SparkSession，涉及到表名、字段名称和字段类型的元数据都将保存在SessionCatalog中；2当调用SparkSession的sql()方法时就会使用SparkSqlParser进行解析SQL语句，解析过程中使用的ANTLR进行词法解析和语法解析；3使用Analyzer分析器绑定逻辑计划，在该阶段，Analyzer会使用AnalyzerRules，并结合SessionCatalog，对未绑定的逻辑计划进行解析，生成已绑定的逻辑计划；SparkSQL的架构4使用Optimizer优化器优化逻辑计划，该优化器同样定义了一套规则（Rules），利用这些规则对逻辑计划和语句进行迭代处理；5使用SparkPlanner对优化后的逻辑计划进行转换，生成可以执行的物理计划SparkPlan；6使用QueryExecution执行物理计划，此时则调用SparkPlan的execute()方法，返回RDDs。SparkSQL工作流程：12小结知道SparkSQL的作用掌握SparkSQL的架构通信数据分析与实战SparkSQL第四章第2节14知道DataFrame和RDD的区别熟悉DataFrame的创建掌握DataFrame的常用操作学习目标TARGETDataFrame的概述SparkSQL使用的数据对象并非是RDD，而是DataFrame。在Spark1.3.0版本之前，DataFrame被称为SchemaRDD。DataFrame使Spark具备处理大规模结构化数据的能力。在Spark中，DataFrame是一种以RDD为基础的分布式数据集。DataFrame的结构类似传统数据库的二维表格，可以从很多数据源中创建，如结构化文件、外部数据库、Hive表等数据源。DataFrame的概述DataFrame可以看作是分布式的Row对象的集合，在二维表数据集的每一列都带有名称和类型，这就是Schema元信息，这使得Spark框架可获取更多数据结构信息，从而对在DataFrame背后的数据源以及作用于DataFrame之上数据变换进行针对性的优化，最终达到提升计算效率。DataFrame的创建在Spark2.0版本之前，SparkSQL中的SQLContext是创建DataFrame和执行SQL的入口,而在Spark2.0之后，Spark使用全新的[SparkSession接口]()替代SQLContext接口完成数据的加载、转换、处理等功能。创建SparkSession对象可以通过“SparkSession.builder().getOrCreate()”方法获取，但当我们使用Spark-Shell编写程序时，Spark-Shell客户端会默认提供了一个名为“sc”的SparkContext对象和一个名为“spark”的SparkSession对象，因此我们可以直接使用这两个对象，不需要自行创建。创建DataFrame的两种基本方式：已存在的RDD调用toDF()方法转换得到DataFrame。通过Spark读取数据源直接创建DataFrame。DataFrame的创建在创建DataFrame之前,为了支持RDD转换成DataFrame及后续的SQL操作,需要导入spark.implicits._包启用隐士转换.若使用SparkSession方式创建DataFrame，可以使用spark.read从不同类型的文件中加载数据创建DataFrame。spark.read的具体操作，如下所示。DataFrame的创建1．数据准备在HDFS文件系统中的/spark目录中有一个person.txt文件，内容如下：

1zhongnanshan852yuanlongping923yumin934shenjilan915sunjiadong926张富清93DataFrame的创建2．通过文件直接创建DataFramescala>valpersonDF=spark.read.text("/spark/person.txt")personDF:org.apache.spark.sql.DataFrame=[value:String]scala>personDF.printSchema()root|--value:String(Nullable=true)我们通过Spark读取数据源的方式进行创建DataFrame

DataFrame的创建3．RDD直接转换为DataFramescala>vallineRDD=sc.textFile("/spark/person.txt").map(_.split(""))lineRDD:org.apache.spark.rdd.RDD[Array[String]]=MapPartitionsRDD[6]atmapat<console>:24scala>caseclassPerson(id:Int,name:String,age:Int)scala>valpersonRDD=lineRDD.map(x=>Person(x(0).toInt,x(1),x(2).toInt))personRDD:org.apache.spark.rdd.RDD[Person]=MapPartitionsRDD[7]atmapat<console>:27scala>valpersonDF=personRDD.toDF()personDF:org.apache.spark.sql.DataFrame=[id:int,name:string...1morefield]DataFrame的常用操作DataFrame提供了两种语法风格，即DSL风格语法和SQL风格语法。二者在功能上并无区别，仅仅是根据用户习惯，自定义选择操作方式。DataFrame提供了一个领域特定语言（DSL）以方便操作结构化数据。在程序中直接使用spark.sql()方式执行SQL查询，结果将作为一个DataFrame返回，使用SQL风格操作的前提是将DataFrame注册成一个临时表。DSLSQLDataFrame的常用操作DSL的风格操作方法名称相关说明show()查看DataFrame中的具体内容信息printSchema()查看DataFrame的Schema信息select()查看DataFrame中选取部分列的数据及进行重命名filter()实现条件查询，过滤出想要的结果groupBy()对记录进行分组sort()对特定字段进行排序操作DataFrame的常用操作SQL的风格操作DataFrame的强大之处就是我们可以将它看作是一个关系型数据表，然后可以在程序中直接使用spark.sql()的方式执行SQL查询，结果将作为一个DataFrame返回。使用SQL风格操作的前提是需要将DataFrame注册成一个临时表1.

将DataFrame注册成一个临时表

scala>personDF.registerTempTable("t_person")DataFrame的常用操作SQL的风格操作2.

查询年龄最大的前两名人的信息scala>spark.sql("select*fromt_personorderbyagedesclimit2").show()3.

查询年龄大于92的人的信息scala>spark.sql("select*fromt_personwhereage>92").show()26小结知道DataFrame和RDD的区别熟悉DataFrame的创建掌握DataFrame的常用操作通信数据分析与实战SparkSQL第四章第3节28熟悉Dataset的数据格式掌握Dataset的创建学习目标TARGETDataSet的简介Dataset是从Spark1.6Alpha版本中引入的一个新的数据抽象结构，最终在Spark2.0版本被定义成Spark新特性。Dataset提供了特定域对象中的强类型集合，也就是在RDD的每行数据中添加了类型约束条件，只有约束条件的数据类型才能正常运行。Dataset结合了RDD和DataFrame的优点，并且可以调用封装的方法以并行方式进行转换等操作。RDD、DataFrame及Dataset的区别RDD数据的表现形式，即序号（1），此时RDD数据没有数据类型和元数据信息。DataFrame数据的表现形式，即序号（2），此时DataFrame数据中添加Schema元数据信息（列名和数据类型，如ID：String），DataFrame每行类型固定为Row类型，每列的值无法直接访问，只有通过解析才能获取各个字段的值。RDD、DataFrame及Dataset的区别Dataset数据的表现形式，序号（3）和（4），其中序号（3）是在RDD每行数据的基础之上，添加一个数据类型（value：String）作为Schema元数据信息。而序号（4）每行数据添加People强数据类型，在Dataset[Person]中里存放了3个字段和属性，Dataset每行数据类型可自定义，一旦定义后，就具有错误检查机制。Dataset的创建创建Dataset可以通过SparkSession中的createDataset来创建scala>valpersonDs=spark.createDataset(sc.textFile("/spark/person.txt"))personDs:org.apache.spark.sql.Dataset[String]=[value:string]scala>personDs.show()Dataset的创建DataFrame通过“as[ElementType]”方法转换得到Datasetscala>spark.read.text("/spark/person.txt").as[String]res14:org.apache.spark.sql.Dataset[String]=[value:string]scala>spark.read.text("/spark/person.txt").as[String].toDF()res15:org.apache.spark.sql.DataFrame=[value:string]34小结熟悉Dataset的数据格式掌握Dataset的创建通信数据分析与实战SparkSQL第四章第4节36掌握利用反射机制RDD转换DataFrame掌握通过编程接口RDD转换DataFrame学习目标TARGETRDD转换DataFrameSpark官方提供了两种方法实现从RDD转换得到DataFrame。第一种方法是利用反射机制来推断包含特定类型对象的Schema，这种方式适用于对已知数据结构的RDD转换第二种方法通过编程接口构造一个Schema，并将其应用在已知的RDD数据中。反射机制编程接口RDD转换DataFrameWindows系统开发Scala代码，可使用本地环境测试（需要先准备本地数据文件）。我们可以很容易的分析出当前数据文件中字段的信息，但计算机无法直观感受字段的实际含义，因此需要通过反射机制来推断包含特定类型对象的Schema信息，实现将RDD转换成DataFrame。1创建Maven工程。打开IDEA开发工具，创建名为“spark_chapter04”的Maven工程。2添加依赖。在pom.xml文件中添加SparkSQL依赖。3定义caseclass样例类、字段和属性，样例类的参数名会被利用反射机制作为列名。通过sc对象读取文件生成一个RDD，将RDD与样例类匹配，调用toDF()方法将RDD转换为DataFrame。RDD转换DataFrame2添加依赖。在pom.xml文件中添加SparkSQL依赖。<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.3.2</version></dependency>RDD转换DataFrame3定义caseclass样例类、字段和属性，样例类的参数名会被利用反射机制作为列名。通过sc对象读取文件生成一个RDD，将RDD与样例类匹配，调用toDF()方法将RDD转换为DataFrame。//定义样例类

caseclassPerson(id:Int,name:String,age:Int)

objectCaseClassSchema{

defmain(args:Array[String]):Unit={

//1.构建SparkSession

valspark:SparkSession=SparkSession.builder()

.appName("CaseClassSchema")

.master("local[2]")

.getOrCreate()

//2.获取SparkContext

valsc:SparkContext=spark.sparkContext

//设置日志打印级别

sc.setLogLevel("WARN")RDD转换DataFrame //3.读取文件

valdata:RDD[Array[String]]=

sc.textFile(“C://data//person.txt").map(x=>x.split(""))

//4.将RDD与样例类关联

valpersonRdd:RDD[Person]=

data.map(x=>Person(x(0).toInt,x(1),x(2).toInt))

//5.获取DataFrame

//手动导入隐式转换

importspark.implicits._

valpersonDF:DataFrame=personRdd.toDF

RDD转换DataFrame

//------------DSL风格操作开始-------------//1.显示DataFrame的数据，默认显示20行

personDF.show()//2.显示DataFrame的schema信息

personDF.printSchema()//3.统计DataFrame中年龄大于30的人数

println(personDF.filter($"age">30).count())

//-----------DSL风格操作结束-------------

//-----------SQL风格操作开始-------------//将DataFrame注册成表

personDF.createOrReplaceTempView("t_person")spark.sql("select*fromt_person").show()spark.sql("select*fromt_personwherename='zhangsan'").show()

//-----------SQL风格操作结束-------------//关闭资源操作

sc.stop()spark.stop()RDD转换DataFrame当Case类不能提前定义Schema时，就需要采用编程方式定义Schema信息，实现RDD转换DataFrame的功能。1创建一个Row对象结构的RDD。2基于StructType类型创建Schema。3通过SparkSession提供的createDataFrame()方法来拼接Schema，来实现RDD转换成DataFrame。RDD转换DataFrame1创建一个Row对象结构的RDD。objectSparkSqlSchema{

defmain(args:Array[String]):Unit={

//1.创建SparkSession

valspark:SparkSession=SparkSession.builder()

.appName("SparkSqlSchema")

.master("local[2]")

.getOrCreate()

//2.获取sparkContext对象

valsc:SparkContext=spark.sparkContext

//设置日志打印级别

sc.setLogLevel("WARN")RDD转换DataFrame//3.加载数据

valdataRDD:RDD[String]=sc.textFile("D://spark//person.txt")

//4.切分每一行

valdataArrayRDD:RDD[Array[String]]=dataRDD.map(_.split(""))

//5.加载数据到Row对象中

valpersonRDD:RDD[Row]=

dataArrayRDD.map(x=>Row(x(0).toInt,x(1),x(2).toInt))

//6.创建Schema

valschema:StructType=StructType(Seq(

StructField("id",IntegerType,false),

StructField("name",StringType,false),

StructField("age",IntegerType,false)

))2基于StructType类型创建Schema。RDD转换DataFrame//7.利用personRDD与Schema创建DataFrame

valpersonDF:DataFrame=spark.createDataFrame(personRDD,schema)

//8.DSL操作显示DataFrame的数据结果

personDF.show()

//9.将DataFrame注册成表

personDF.createOrReplaceTempView("t_person")

//10.sql语句操作

spark.sql("select*fromt_person").show()

//11.关闭资源

sc.stop()

spark.stop()3通过SparkSession提供的createDataFrame()方法来拼接Schema，来实现RDD转换成DataFrame。47小结掌握利用反射机制RDD转换DataFrame掌握通过编程接口RDD转换DataFrame通信数据分析与实战SparkSQL第四章第5节49掌握SparkSQL操作MySQL掌握SparkSQL操作Hive学习目标TARGETSparkSQL操作Mysql1创建并读取MySQL数据库。2在IDEA中添加MySQL驱动连接包

3读取MySQL数据库中的数据。4向MySQL数据库写入数据

。操作步骤：

SparkSQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame进行一系列的操作后，还可以将数据重新写入到关系型数据中。关于SparkSQL对MySQL数据库的相关操作具体如下:SparkSQL操作Mysql1创建并读取MySQL数据库。在hadoop01节点上使用MySQL客户端创建数据库、数据表以及插入数据，具体命令如下。#启动mysql客户端mysql-uroot–p123456#创建spark数据库mysql>CREATEdatabasespark;ps–ef|grepmysql#确认Mysql服务是否开启/etc/init.d/mysqldstart#如果没有开启,需要主动开启服务SparkSQL操作Mysql1创建并读取MySQL数据库。在hadoop01节点上使用MySQL客户端创建数据库、数据表以及插入数据，具体命令如下。#创建person数据表

mysql>CREATETABLEperson(idINT(4),NAMECHAR(20),ageINT(4));#插入数据mysql>INSERTINTOpersonVALUE(1,‘tuyouyou’,91);mysql>INSERTINTOpersonVALUE(2,’yuanlongping’,91);mysql>SELECT*FROMperson;SparkSQL操作Mysql2在IDEA中添加MySQL驱动连接包

pom.xml<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.38</version></dependency>SparkSQL操作Mysql3读取MySQL中的数据importjava.util.Propertiesimportorg.apache.spark.sql.{DataFrame,SparkSession}objectDataFromMysql{

defmain(args:Array[String]):Unit={

//1、创建sparkSession对象

valspark:SparkSession=SparkSession.builder().appName("DataFromMysql").master("local[2]").getOrCreate()

//2、创建Properties对象，设置连接mysql的用户名和密码

valproperties:Properties=newProperties()

properties.setProperty("user","root")

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

//3、读取mysql中的数据

valmysqlDF:DataFrame=spark.read.jdbc

("jdbc:mysql://34:3306/spark","person",properties)

//4、显示mysql中表的数据

mysqlDF.show()

spark.stop()

SparkSQL操作Mysql4写入MySQL中的数据

importjava.util.Propertiesimportorg.apache.spark.rdd.RDDimportorg.apache.spark.sql.{DataFrame,SparkSession}

//创建样例类PersoncaseclassPerson(id:Int,name:String,age:Int)

objectSparkSqlToMysql{

defmain(args:Array[String]):Unit={

//1.创建sparkSession对象

valspark:SparkSession=SparkSession.builder()

.appName("SparkSqlToMysql")

.master("local[2]")

.getOrCreate()

//2.创建数据

valdata=spark.sparkContext

.parallelize(Array(“3,shenjilan,91","4,yumin,93"))

//3.按MySQL列名切分数据

valarrRDD:RDD[Array[String]]=data.map(_.split(","))SparkSQL操作Mysql4写入MySQL中的数据

//4.RDD关联Person样例类

valpersonRDD:RDD[Person]=

arrRDD.map(x=>Person(x(0).toInt,x(1),x(2).toInt))

//导入隐式转换

importspark.implicits._

//5.将RDD转换成DataFrame

valpersonDF:DataFrame=personRDD.toDF()

//6.设置JDBC配置参数

valprop=newProperties()

prop.setProperty("user","root")

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

prop.setProperty("driver","com.mysql.jdbc.Driver")

//7.写入数据

personDF.write.mode("append").jdbc(

"jdbc:mysql://34:3306/spark","spark.person",prop)

personDF.show()SparkSQL操作Hive

ApacheHive是Hadoop上的SQL引擎，也是大数据系统中重要的数据仓库工具，SparkSQL支持访问Hive数据仓库，

然后在Spark引擎中进行统计分析。接下来介绍通过SparkSQL操作Hive数据仓库的具体实现步骤SparkSQL操作Hive1准备环境，使Spark可以访问Hive，SparkSQL连接Hive。2在Hive中创建数据库和表。3SparkSQL操作Hive数据库。4向Hive表写入数据。操作步骤：SparkSQL操作Hive1准备环境，使Spark可以访问Hive，SparkSQL连接Hive。Hive采用MySQL数据库存放Hive元数据，因此为了能够让Spark访问Hive，就需要将MySQL驱动包拷贝到Spark安装路径下的Jars目录下，具体命令如下:要把SparkSQL连接到一个部署好的Hive时，就必须要把hive-site.xml配置文件复制到Spark的配置文件目录中，这里采用软连接方式，具体命令如下:cd/export/servers/apache-hive-1.2.1-bin/lib/cpmysql-connector-java-5.1.32.jar/export/servers/spark/jars/ln-s/export/servers/apache-hive-1.2.1-bin/conf/hive-site.xml/export/servers/spark/conf/hive-site.xmlSparkSQL操作Hive2在Hive中创建数据库和表接下来，我们首先在hadoop01节点上启动Hive服务，创建数据库和表，具体命令如下所示。#启动hive程序cd/export/servers/apache-hive-1.2.1-bin/bin/

hive#创建数据仓库

hive>createdatabasesparksqltest;#创建数据表hive>createtableif