《通信數據分析與實戰(zhàn)》課件-第四章 Spark SQL

通信數據分析與實戰(zhàn)SparkSQL第四章第1節(jié)2知道SparkSQL的作用掌握SparkSQL的架構學習目標TARGETSparkSQL的概述在很多情況下，開發(fā)工程師并不了解Scala語言，也不了解Spark常用API，但又非常想要使用Spark框架提供的強大的數據分析能力。Spark的開發(fā)工程師們考慮到了這個問題，利用SQL語言的語法簡潔、學習門檻低以及在編程語言普及程度和流行程度高等諸多優(yōu)勢，從而開發(fā)了SparkSQL模塊，通過SparkSQL，開發(fā)人員能夠通過使用SQL語句，實現對結構化數據的處理。SparkSQL的概述SparkSQL是Spark用來處理結構化數據的一個模塊，它提供了一個編程抽象結構叫做DataFrame的數據模型（即帶有Schema信息的RDD），SparkSQL作為分布式SQL查詢引擎，讓用戶可以通過SQL、DataFramesAPI和DatasetsAPI三種方式實現對結構化數據的處理。SparkSQL的概述SparkSQL主要提供了以下三個功能：SparkSQL可從各種結構化數據源中讀取數據，進行數據分析。SparkSQL包含行業(yè)標準的JDBC和ODBC連接方式，因此它不局限于在Spark程序內使用SQL語句進行查詢。SparkSQL可以無縫地將SQL查詢與Spark程序進行結合，它能夠將結構化數據作為Spark中的分布式數據集（RDD）進行查詢。SparkSQL的架構SparkSQL架構與Hive架構相比，把底層的MapReduce執(zhí)行引擎更改為Spark，還修改了Catalyst優(yōu)化器，SparkSQL快速的計算效率得益于Catalyst優(yōu)化器。從HiveQL被解析成語法抽象樹起，執(zhí)行計劃生成和優(yōu)化的工作全部交給SparkSQL的Catalyst優(yōu)化器進行負責和管理。SparkSQL架構圖SparkSQL的架構Spark要想很好地支持SQL，需要完成解析（Parser）、優(yōu)化（Optimizer）、執(zhí)行（Execution）三大過程。SparkSQL的架構SqlParse：完成SQL語法解析功能，目前只提供了一個簡單的SQL解析器。Analyze：主要完成綁定工作，將不同來源的UnresolvedLogicalPlan和元數據進行綁定，生成ResolvedLogicalPlan。Optimizer：對ResolvedLo;gicalPlan進行優(yōu)化，生成OptimizedLogicalPlan。Planner：將LogicalPlan轉換成PhysicalPlan。CostModel：主要根據過去的性能統計數據，選擇最佳的物理執(zhí)行計劃。Catalyst優(yōu)化器在執(zhí)行計劃生成和優(yōu)化的工作時，離不開內部的五大組件。SparkSQL的架構在了解了上述組件的作用后，下面分步驟講解SparkSQL工作流程。

SparkSQL的架構SparkSQL工作流程：1下在解析SQL語句之前，會創(chuàng)建SparkSession，涉及到表名、字段名稱和字段類型的元數據都將保存在SessionCatalog中；2當調用SparkSession的sql()方法時就會使用SparkSqlParser進行解析SQL語句，解析過程中使用的ANTLR進行詞法解析和語法解析；3使用Analyzer分析器綁定邏輯計劃，在該階段，Analyzer會使用AnalyzerRules，并結合SessionCatalog，對未綁定的邏輯計劃進行解析，生成已綁定的邏輯計劃；SparkSQL的架構4使用Optimizer優(yōu)化器優(yōu)化邏輯計劃，該優(yōu)化器同樣定義了一套規(guī)則（Rules），利用這些規(guī)則對邏輯計劃和語句進行迭代處理；5使用SparkPlanner對優(yōu)化后的邏輯計劃進行轉換，生成可以執(zhí)行的物理計劃SparkPlan；6使用QueryExecution執(zhí)行物理計劃，此時則調用SparkPlan的execute()方法，返回RDDs。SparkSQL工作流程：12小結知道SparkSQL的作用掌握SparkSQL的架構通信數據分析與實戰(zhàn)SparkSQL第四章第2節(jié)14知道DataFrame和RDD的區(qū)別熟悉DataFrame的創(chuàng)建掌握DataFrame的常用操作學習目標TARGETDataFrame的概述SparkSQL使用的數據對象并非是RDD，而是DataFrame。在Spark1.3.0版本之前，DataFrame被稱為SchemaRDD。DataFrame使Spark具備處理大規(guī)模結構化數據的能力。在Spark中，DataFrame是一種以RDD為基礎的分布式數據集。DataFrame的結構類似傳統數據庫的二維表格，可以從很多數據源中創(chuàng)建，如結構化文件、外部數據庫、Hive表等數據源。DataFrame的概述DataFrame可以看作是分布式的Row對象的集合，在二維表數據集的每一列都帶有名稱和類型，這就是Schema元信息，這使得Spark框架可獲取更多數據結構信息，從而對在DataFrame背后的數據源以及作用于DataFrame之上數據變換進行針對性的優(yōu)化，最終達到提升計算效率。DataFrame的創(chuàng)建在Spark2.0版本之前，SparkSQL中的SQLContext是創(chuàng)建DataFrame和執(zhí)行SQL的入口,而在Spark2.0之后，Spark使用全新的[SparkSession接口]()替代SQLContext接口完成數據的加載、轉換、處理等功能。創(chuàng)建SparkSession對象可以通過“SparkSession.builder().getOrCreate()”方法獲取，但當我們使用Spark-Shell編寫程序時，Spark-Shell客戶端會默認提供了一個名為“sc”的SparkContext對象和一個名為“spark”的SparkSession對象，因此我們可以直接使用這兩個對象，不需要自行創(chuàng)建。創(chuàng)建DataFrame的兩種基本方式：已存在的RDD調用toDF()方法轉換得到DataFrame。通過Spark讀取數據源直接創(chuàng)建DataFrame。DataFrame的創(chuàng)建在創(chuàng)建DataFrame之前,為了支持RDD轉換成DataFrame及后續(xù)的SQL操作,需要導入spark.implicits._包啟用隱士轉換.若使用SparkSession方式創(chuàng)建DataFrame，可以使用spark.read從不同類型的文件中加載數據創(chuàng)建DataFrame。spark.read的具體操作，如下所示。DataFrame的創(chuàng)建1．數據準備在HDFS文件系統中的/spark目錄中有一個person.txt文件，內容如下：

1zhongnanshan852yuanlongping923yumin934shenjilan915sunjiadong926張富清93DataFrame的創(chuàng)建2．通過文件直接創(chuàng)建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讀取數據源的方式進行創(chuàng)建DataFrame

DataFrame的創(chuàng)建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風格語法。二者在功能上并無區(qū)別，僅僅是根據用戶習慣，自定義選擇操作方式。DataFrame提供了一個領域特定語言（DSL）以方便操作結構化數據。在程序中直接使用spark.sql()方式執(zhí)行SQL查詢，結果將作為一個DataFrame返回，使用SQL風格操作的前提是將DataFrame注冊成一個臨時表。DSLSQLDataFrame的常用操作DSL的風格操作方法名稱相關說明show()查看DataFrame中的具體內容信息printSchema()查看DataFrame的Schema信息select()查看DataFrame中選取部分列的數據及進行重命名filter()實現條件查詢，過濾出想要的結果groupBy()對記錄進行分組sort()對特定字段進行排序操作DataFrame的常用操作SQL的風格操作DataFrame的強大之處就是我們可以將它看作是一個關系型數據表，然后可以在程序中直接使用spark.sql()的方式執(zhí)行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的區(qū)別熟悉DataFrame的創(chuàng)建掌握DataFrame的常用操作通信數據分析與實戰(zhàn)SparkSQL第四章第3節(jié)28熟悉Dataset的數據格式掌握Dataset的創(chuàng)建學習目標TARGETDataSet的簡介Dataset是從Spark1.6Alpha版本中引入的一個新的數據抽象結構，最終在Spark2.0版本被定義成Spark新特性。Dataset提供了特定域對象中的強類型集合，也就是在RDD的每行數據中添加了類型約束條件，只有約束條件的數據類型才能正常運行。Dataset結合了RDD和DataFrame的優(yōu)點，并且可以調用封裝的方法以并行方式進行轉換等操作。RDD、DataFrame及Dataset的區(qū)別RDD數據的表現形式，即序號（1），此時RDD數據沒有數據類型和元數據信息。DataFrame數據的表現形式，即序號（2），此時DataFrame數據中添加Schema元數據信息（列名和數據類型，如ID：String），DataFrame每行類型固定為Row類型，每列的值無法直接訪問，只有通過解析才能獲取各個字段的值。RDD、DataFrame及Dataset的區(qū)別Dataset數據的表現形式，序號（3）和（4），其中序號（3）是在RDD每行數據的基礎之上，添加一個數據類型（value：String）作為Schema元數據信息。而序號（4）每行數據添加People強數據類型，在Dataset[Person]中里存放了3個字段和屬性，Dataset每行數據類型可自定義，一旦定義后，就具有錯誤檢查機制。Dataset的創(chuàng)建創(chuàng)建Dataset可以通過SparkSession中的createDataset來創(chuàng)建scala>valpersonDs=spark.createDataset(sc.textFile("/spark/person.txt"))personDs:org.apache.spark.sql.Dataset[String]=[value:string]scala>personDs.show()Dataset的創(chuàng)建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的創(chuàng)建通信數據分析與實戰(zhàn)SparkSQL第四章第4節(jié)36掌握利用反射機制RDD轉換DataFrame掌握通過編程接口RDD轉換DataFrame學習目標TARGETRDD轉換DataFrameSpark官方提供了兩種方法實現從RDD轉換得到DataFrame。第一種方法是利用反射機制來推斷包含特定類型對象的Schema，這種方式適用于對已知數據結構的RDD轉換第二種方法通過編程接口構造一個Schema，并將其應用在已知的RDD數據中。反射機制編程接口RDD轉換DataFrameWindows系統開發(fā)Scala代碼，可使用本地環(huán)境測試（需要先準備本地數據文件）。我們可以很容易的分析出當前數據文件中字段的信息，但計算機無法直觀感受字段的實際含義，因此需要通過反射機制來推斷包含特定類型對象的Schema信息，實現將RDD轉換成DataFrame。1創(chuàng)建Maven工程。打開IDEA開發(fā)工具，創(chuàng)建名為“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創(chuàng)建一個Row對象結構的RDD。2基于StructType類型創(chuàng)建Schema。3通過SparkSession提供的createDataFrame()方法來拼接Schema，來實現RDD轉換成DataFrame。RDD轉換DataFrame1創(chuàng)建一個Row對象結構的RDD。objectSparkSqlSchema{

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

//1.創(chuàng)建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.創(chuàng)建Schema

valschema:StructType=StructType(Seq(

StructField("id",IntegerType,false),

StructField("name",StringType,false),

StructField("age",IntegerType,false)

))2基于StructType類型創(chuàng)建Schema。RDD轉換DataFrame//7.利用personRDD與Schema創(chuàng)建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通信數據分析與實戰(zhàn)SparkSQL第四章第5節(jié)49掌握SparkSQL操作MySQL掌握SparkSQL操作Hive學習目標TARGETSparkSQL操作Mysql1創(chuàng)建并讀取MySQL數據庫。2在IDEA中添加MySQL驅動連接包

3讀取MySQL數據庫中的數據。4向MySQL數據庫寫入數據

。操作步驟：

SparkSQL可以通過JDBC從關系型數據庫中讀取數據的方式創(chuàng)建DataFrame，通過對DataFrame進行一系列的操作后，還可以將數據重新寫入到關系型數據中。關于SparkSQL對MySQL數據庫的相關操作具體如下:SparkSQL操作Mysql1創(chuàng)建并讀取MySQL數據庫。在hadoop01節(jié)點上使用MySQL客戶端創(chuàng)建數據庫、數據表以及插入數據，具體命令如下。#啟動mysql客戶端mysql-uroot–p123456#創(chuàng)建spark數據庫mysql>CREATEdatabasespark;ps–ef|grepmysql#確認Mysql服務是否開啟/etc/init.d/mysqldstart#如果沒有開啟,需要主動開啟服務SparkSQL操作Mysql1創(chuàng)建并讀取MySQL數據庫。在hadoop01節(jié)點上使用MySQL客戶端創(chuàng)建數據庫、數據表以及插入數據，具體命令如下。#創(chuàng)建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、創(chuàng)建sparkSession對象

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

//2、創(chuàng)建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}

//創(chuàng)建樣例類PersoncaseclassPerson(id:Int,name:String,age:Int)

objectSparkSqlToMysql{

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

//1.創(chuàng)建sparkSession對象

valspark:SparkSession=SparkSession.builder()

.appName("SparkSqlToMysql")

.master("local[2]")

.getOrCreate()

//2.創(chuàng)建數據

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準備環(huán)境，使Spark可以訪問Hive，SparkSQL連接Hive。2在Hive中創(chuàng)建數據庫和表。3SparkSQL操作Hive數據庫。4向Hive表寫入數據。操作步驟：SparkSQL操作Hive1準備環(huán)境，使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中創(chuàng)建數據庫和表接下來，我們首先在hadoop01節(jié)點上啟動Hive服務，創(chuàng)建數據庫和表，具體命令如下所示。#啟動hive程序cd/export/servers/apache-hive-1.2.1-bin/bin/

hive#創(chuàng)建數據倉庫

hive>createdatabasesparksqltest;#創(chuàng)建數據表hive>createtableif