第5章 Spark SQL-結(jié)構(gòu)化數(shù)據(jù)文件處理

SparkSQL

——結(jié)構(gòu)化數(shù)據(jù)文件處理基于大數(shù)據(jù)技術(shù)對(duì)房?jī)r(jià)進(jìn)行分析和預(yù)測(cè)，是科學(xué)精準(zhǔn)調(diào)控，促進(jìn)房地產(chǎn)市場(chǎng)平穩(wěn)健康發(fā)展，實(shí)現(xiàn)“房住不炒”的重要手段?，F(xiàn)有一份房屋銷售數(shù)據(jù)文件house.csv，記錄了某地區(qū)的房屋銷售情況，包含銷售價(jià)格、銷售日期、房屋評(píng)分等共14個(gè)數(shù)據(jù)字段，字段說(shuō)明如下。（1英尺=0.3048米）任務(wù)背景字段名稱說(shuō)明

字段名稱說(shuō)明selling_price銷售價(jià)格（單位：美元）built_area建筑面積（單位：平方英尺）bedrooms_num臥室數(shù)basement_area地下室面積（單位：平方英尺）bathroom_num浴室數(shù)year_bulit修建年份housing_area房屋面積（單位：平方英尺）year_repair修復(fù)年份parking_area停車區(qū)面積（單位：平方英尺）latitude緯度f(wàn)loor_num樓層數(shù)longitude經(jīng)度sales_data銷售日期housing_rating房屋評(píng)分在進(jìn)行房?jī)r(jià)數(shù)據(jù)分析之前，由于無(wú)法直接判斷出各個(gè)數(shù)據(jù)字段之間的關(guān)系，因此需要先對(duì)數(shù)據(jù)進(jìn)行基礎(chǔ)的探索，探索各個(gè)數(shù)據(jù)字段間的關(guān)系并加以分析。本章將使用SparkSQL即席查詢框架解決房?jī)r(jià)數(shù)據(jù)探索分析的問題。首先介紹SparkSQL框架及其編程數(shù)據(jù)模型DataFrame的基本概念，并對(duì)SparkSQL相關(guān)環(huán)境進(jìn)行配置。其次介紹DataFrame的查詢、輸出操作API的用法。最后結(jié)合房?jī)r(jià)數(shù)據(jù)探索分析實(shí)例，幫助讀者掌握DataFrame的基礎(chǔ)操作。任務(wù)背景1掌握DataFrame基礎(chǔ)操作目錄認(rèn)識(shí)SparkSQL2探索分析房屋售價(jià)數(shù)據(jù)3使用SparkSQL探索分析房?jī)r(jià)數(shù)據(jù)前，需要先了解SparkSQL是什么，有什么作用。本節(jié)的任務(wù)如下。了解SparkSQL框架、SparkSQL的編程模型DataFrame和SparkSQL的運(yùn)行過程。對(duì)SparkSQL相關(guān)環(huán)境進(jìn)行配置。任務(wù)描述SparkSQL是一個(gè)用來(lái)處理結(jié)構(gòu)化數(shù)據(jù)的Spark框架可被視為一個(gè)分布式的SQL查詢引擎，并且提供了一個(gè)抽象的可編程數(shù)據(jù)模型DataFrame。SparkSQL可以直接處理RDD，也可以處理Parquet文件或者JSON文件，甚至可以處理外部數(shù)據(jù)庫(kù)中的數(shù)據(jù)以及Hive中存在的表數(shù)據(jù)。了解SparkSQL基本概念SparkSQL架構(gòu)了解SparkSQL基本概念將hive-site.xml復(fù)制到$SPARK_HOME/conf目錄下。在$SPARK_HOME/conf/spark-env.sh文件中配置MySQL驅(qū)動(dòng)。啟動(dòng)MySQL服務(wù)。啟動(dòng)Hive的metastore服務(wù)。修改日志級(jí)別。啟動(dòng)Spark集群。啟動(dòng)spark-sql。配置SparkSQL啟動(dòng)spark-shell即可使用SparkSQL的Shell交互接口。如果在spark-shell中執(zhí)行SQL語(yǔ)句，需要使用SparkSession對(duì)象來(lái)調(diào)用sql()方法。HiveContext現(xiàn)在不僅支持HiveQL語(yǔ)法解析器，同時(shí)也支持SQL語(yǔ)法解析器。在spark-shell啟動(dòng)的過程中會(huì)初始化SparkSession對(duì)象為spark，此時(shí)初始化的spark對(duì)象既支持SQL語(yǔ)法解析器，也支持HiveQL語(yǔ)法解析器。也就是使用這個(gè)spark可以執(zhí)行SQL語(yǔ)句和HQL語(yǔ)句。如果需要支持Hive，還需要啟用enableHiveSupport()方法，并且Hive的配置文件hive-site.xml已經(jīng)存在于工程中。了解SparkSQL與Shell交互1掌握DataFrame基礎(chǔ)操作目錄認(rèn)識(shí)SparkSQL2探索分析房屋售價(jià)數(shù)據(jù)3SparkSQL提供了一個(gè)抽象的編程數(shù)據(jù)模型DataFrame，DataFrame是由SchemaRDD發(fā)展而來(lái)的，從Spark1.3.0開始，SchemaRDD更名為DataFrame。SchemaRDD直接繼承自RDD，而DataFrame則自身實(shí)現(xiàn)RDD的絕大多數(shù)功能?？梢詫parkSQL的DataFrame理解為一個(gè)分布式的Row對(duì)象的數(shù)據(jù)集合，該數(shù)據(jù)集合提供了由列組成的詳細(xì)模式信息。本節(jié)的任務(wù)是學(xué)習(xí)DataFrame對(duì)象的創(chuàng)建方法及基礎(chǔ)的操作。任務(wù)描述通過結(jié)構(gòu)化數(shù)據(jù)文件創(chuàng)建DataFrame通過外部數(shù)據(jù)庫(kù)創(chuàng)建DataFrame通過RDD創(chuàng)建DataFrame通過Hive中的表創(chuàng)建DataFrame創(chuàng)建DataFrame對(duì)象一般情況下，結(jié)構(gòu)化數(shù)據(jù)文件存儲(chǔ)在HDFS中，較為常見的結(jié)構(gòu)化數(shù)據(jù)文件是Parquet文件或JSON文件。SparkSQL可以通過load()方法將HDFS上的結(jié)構(gòu)化文件數(shù)據(jù)轉(zhuǎn)換為DataFrame，load()方法默認(rèn)導(dǎo)入的文件格式是Parquet。若加載JSON格式的文件數(shù)據(jù)，將其轉(zhuǎn)換為DataFrame，則還需要使用format()方法。也可以直接使用json()方法將JSON文件數(shù)據(jù)轉(zhuǎn)換為DataFrame。創(chuàng)建DataFrame對(duì)象1.通過結(jié)構(gòu)化數(shù)據(jù)文件創(chuàng)建DataFrameSparkSQL可以從外部數(shù)據(jù)庫(kù)（比如MySQL、Oracle等數(shù)據(jù)庫(kù)）中創(chuàng)建DataFrame。使用這種方式創(chuàng)建DataFrame需要通過JDBC連接或ODBC連接的方式訪問數(shù)據(jù)庫(kù)。讀取MySQL數(shù)據(jù)庫(kù)的表數(shù)據(jù)創(chuàng)建DataFrame對(duì)象2.通過外部數(shù)據(jù)庫(kù)創(chuàng)建DataFrame利用反射機(jī)制推斷RDD模式，使用這種方式首先需要定義一個(gè)caseclass，因?yàn)橹挥衏aseclass才能被Spark隱式地轉(zhuǎn)換為DataFrame。創(chuàng)建DataFrame對(duì)象3.通過RDD創(chuàng)建DataFrame采用編程指定Schema的方式將RDD轉(zhuǎn)換成DataFrame。加載數(shù)據(jù)創(chuàng)建RDD。使用StructType創(chuàng)建一個(gè)和步驟（1）的RDD中的數(shù)據(jù)結(jié)構(gòu)相匹配的Schema。通過createDataFrame()方法將Schema應(yīng)用到RDD上，將RDD數(shù)據(jù)轉(zhuǎn)換成DataFrame。Spark

DataFrame基礎(chǔ)操作使用SparkSession對(duì)象并調(diào)用sql()方法查詢Hive中的表數(shù)據(jù)并將其轉(zhuǎn)換成DataFrame。創(chuàng)建DataFrame對(duì)象4.通過Hive中的表創(chuàng)建DataFrame查看及獲取數(shù)據(jù)的常用函數(shù)或方法將movies.dat電影數(shù)據(jù)上傳至HDFS中，加載數(shù)據(jù)為RDD并將其轉(zhuǎn)換為DataFrame.查看DataFrame數(shù)據(jù)方法描述printSchema打印數(shù)據(jù)模式show查看數(shù)據(jù)first/head/take/takeAsList獲取若干行數(shù)據(jù)collect/collectAsList獲取所有數(shù)據(jù)printSchema函數(shù)查看數(shù)據(jù)模式，打印出列的名稱和類型查看DataFrame數(shù)據(jù)1.printSchema：輸出數(shù)據(jù)模式方法解釋show()顯示前20條記錄show(numRows:Int)顯示numRows條show(truncate:Boolean)是否最多只顯示20個(gè)字符，默認(rèn)為trueshow(numRows:Int,truncate:Boolean)顯示numRows條記錄并設(shè)置過長(zhǎng)字符串的顯示格式查看DataFrame數(shù)據(jù)2.show()：查看數(shù)據(jù)show()方法與show(true)方法一樣，只顯示前20條記錄并且最多只顯示20個(gè)字符若是要顯示所有字符，需要使用show(false)方法查看DataFrame數(shù)據(jù)show(numRows:Int)查看前n行記錄查看DataFrame數(shù)據(jù)方法解釋first獲取第一行記錄head(n:Int)獲取前n行記錄take(n:Int)獲取前n行記錄takeAsList(n:Int)獲取前n行數(shù)據(jù)，并以列表的形式展現(xiàn)查看DataFrame數(shù)據(jù)3．first()/head()/take()/takeAsList()：獲取若干條記錄collect方法可以將DataFrame中的所有數(shù)據(jù)都獲取到，并返回一個(gè)數(shù)組。collectAsList方法可以獲取所有數(shù)據(jù)，返回一個(gè)列表。查看DataFrame數(shù)據(jù)4．collect()/collectAsList()：獲取所有數(shù)據(jù)將DataFrame注冊(cè)成為臨時(shí)表，然后通過SQL語(yǔ)句進(jìn)行查詢掌握DataFrame查詢操作直接在DataFrame對(duì)象上進(jìn)行查詢，DataFrame提供了很多查詢的方法掌握DataFrame查詢操作方法描述where條件查詢select/selectExpr/col/apply查詢指定字段的數(shù)據(jù)信息limit查詢前n行記錄orderby排序查詢groupby分組查詢join連接查詢（1）where()方法DataFrame可以使用where(conditionExpr:String)根據(jù)指定條件進(jìn)行查詢參數(shù)中可以使用and或or該方法的返回結(jié)果仍然為DataFrame類型掌握DataFrame查詢操作1．where()/filter()方法（2）filter()方法DataFrame還可使用filter篩選符合條件的數(shù)據(jù)掌握DataFrame查詢操作（1）select()方法：獲取指定字段值select方法根據(jù)傳入的string類型字段名，獲取指定字段的值，以DataFrame類型返回掌握DataFrame查詢操作2．select()/selectExpr()/col()/apply()方法（2）selectExpr()方法：對(duì)指定字段進(jìn)行特殊處理selectExpr：對(duì)指定字段進(jìn)行特殊處理，可以對(duì)指定字段調(diào)用UDF函數(shù)或者指定別名。selectExpr傳入String類型的參數(shù)，返回DataFrame對(duì)象。掌握DataFrame查詢操作（3）col()/apply()方法col或者apply也可以獲取DataFrame指定字段col或者apply只能獲取一個(gè)字段，并且返回對(duì)象為Column類型掌握DataFrame查詢操作limit方法獲取指定DataFrame的前n行記錄，得到一個(gè)新的DataFrame對(duì)象不同于take與head，limit方法不是Action操作。掌握DataFrame查詢操作3．limit()方法orderBy方法是根據(jù)指定字段排序，默認(rèn)為升序排序若是要求降序排序，可以使用desc(“字段名稱”)或者$”字段名”.desc或者在指定字段前面加“-”來(lái)表示降序排序掌握DataFrame查詢操作4．orderBy()/sort()方法sort方法與orderBy方法一樣，也是根據(jù)指定字段排序，用法也與orderBy一樣掌握DataFrame查詢操作groupBy方法是根據(jù)字段進(jìn)行分組操作groupBy方法有兩種調(diào)用方式，可以傳入String類型的字段名，也可傳入Column類型的對(duì)象。掌握DataFrame查詢操作5．groupBy()方法groupBy方法返回的是GroupedData對(duì)象，GroupedData的操作方法如下表。掌握DataFrame查詢操作方法描述max(colNames:String)獲取分組中指定字段或者所有的數(shù)值類型字段的最大值min(colNames:String)獲取分組中指定字段或者所有的數(shù)字類型字段的最小值mean(colNames:String)獲取分組中指定字段或者所有的數(shù)字類型字段的平均值sum(colNames:String)獲取分組中指定字段或者所有的數(shù)字類型字段和值count()獲取分組中的元素個(gè)數(shù)DataFrame提供了三種join方法用于連接兩個(gè)表掌握DataFrame查詢操作6．join()方法方法描述join(right:DataFrame)兩個(gè)表做笛卡爾積join(right:DataFrame,joinExprs:Column)根據(jù)兩表中相同的某個(gè)字段進(jìn)行連接join(right:DataFrame,joinExprs:Column,joinType:String)根據(jù)兩表中相同的某個(gè)字段進(jìn)行連接并指定連接類型join(right:DataFrame)掌握DataFrame查詢操作join(right:DataFrame,joinExprs:Column)掌握DataFrame查詢操作join(right:DataFrame,joinExprs:Column,joinType:String)連接類型joinType只能是inner、outer、left_outer、right_outer、semijoin中的一種掌握DataFrame查詢操作save()方法可以將DataFrame數(shù)據(jù)保存成文件。saveAsTable()方法可以將DataFrame數(shù)據(jù)保存成持久化的表，并在Hive的元數(shù)據(jù)庫(kù)中創(chuàng)建一個(gè)指針指向該表的位置，持久化的表會(huì)一直保留，即使Spark程序重啟也沒有影響，只要連接至同一個(gè)元數(shù)據(jù)服務(wù)即可讀取表數(shù)據(jù)。讀取持久化表時(shí)，只需要用表名作為參數(shù)，調(diào)用spark.table()方法方法即可得到對(duì)應(yīng)DataFrame。默認(rèn)情況下，saveAsTable()方法會(huì)創(chuàng)建一個(gè)內(nèi)部表，表數(shù)據(jù)的位置是由元數(shù)據(jù)服務(wù)控制的。如果刪除表，那么表數(shù)據(jù)也會(huì)同步刪除。掌握DataFrame輸出操作將DataFrame數(shù)據(jù)保存為文件，實(shí)現(xiàn)步驟如下。首先創(chuàng)建一個(gè)映射對(duì)象，用于存儲(chǔ)save()方法需要用到的數(shù)據(jù)，這里將指定文件的頭信息及文件的保存路徑。從user數(shù)據(jù)中選擇出userId、gender和age這3列字段的數(shù)據(jù)。調(diào)用save()方法將步驟（2）中的DataFrame數(shù)據(jù)保存至copyOfUser.json文件夾中。在HDFS的/user/root/sparkSql目錄下查看保存結(jié)果。掌握DataFrame輸出操作使用saveAsTable()方法將DataFrame對(duì)象copyOfUser保存為copyUser表。掌握DataFrame輸出操作1掌握DataFrame基礎(chǔ)操作目錄認(rèn)識(shí)SparkSQL2探索分析房屋售價(jià)數(shù)據(jù)3從購(gòu)房者角度出發(fā)，影響用戶購(gòu)房的主要因素有房屋價(jià)格、房屋屬性。從地產(chǎn)公司角度出發(fā)，除了房屋本身的居住屬性外，地產(chǎn)公司也會(huì)考慮什么樣的房屋屬性在市場(chǎng)上更具有價(jià)值。本節(jié)的任務(wù)如下。使用SparkSQL實(shí)現(xiàn)房?jī)r(jià)數(shù)據(jù)的探索分析，從數(shù)據(jù)中的房屋售價(jià)、房屋評(píng)分和房屋銷售日期等維度進(jìn)行探索分析，獲得房?jī)r(jià)波動(dòng)的內(nèi)在原因。任務(wù)描述在Hive中創(chuàng)建數(shù)據(jù)庫(kù)house。在house數(shù)據(jù)庫(kù)下創(chuàng)建king_county_house表。將數(shù)據(jù)導(dǎo)入表king_county_house中。通過spark.table()方法讀取Hive中king_county_house表的數(shù)據(jù)。獲取數(shù)據(jù)定義一個(gè)函數(shù)null_count，函數(shù)中使用na()方法可以統(tǒng)計(jì)數(shù)據(jù)每個(gè)字段的缺失值。字段值分布探索的代碼將封裝在max_min_mean_std函數(shù)中，使用selectExpr()方法查詢指定字段的信息，結(jié)合max()、min()、mean()及stddev()方法，統(tǒng)計(jì)指定字段中的最大值、最小值、平均值及標(biāo)準(zhǔn)差。數(shù)據(jù)中的sale_data、year_built和year_rapair字段的含義均表示時(shí)間，進(jìn)行字段值分布的探索并無(wú)實(shí)際意義，因此將使用for循環(huán)遍歷數(shù)據(jù)所有字段，并使用ifelse語(yǔ)句進(jìn)行篩選。探索字段值分布字段值分布和缺失值數(shù)量情況探索字段值分布字段名稱最大值最小值平均值標(biāo)準(zhǔn)差值缺失值數(shù)量sale_data\

\\\1selling_price688500075000542874.928372925.7651bedrooms_num1003.3680.8931bathroom_num7.7502.1170.7741housing_area98903902082.488922.8791parking_area165135957215352.73445776.2291floor_num3.511.5020.5441housing_rating1337.6651.1741built_area88603901791.475829.4491basement_area48200291.014446.6411year_built\\\\1year_repair\\\\1latitude47.777647.159347.5600.1381longitude-121.315-122.519-122.2150.1391由于原始數(shù)據(jù)表中銷售時(shí)間字段“sale_data”為string類型數(shù)據(jù)，所以需要先將該字段轉(zhuǎn)換為日期格式，并通過withColumn()方法在原數(shù)據(jù)上新增一列轉(zhuǎn)換后的字段date。使用quarter()方法將時(shí)間劃分為4個(gè)季度，并使用withColumn()方法將劃分結(jié)果存放在新的字段quarter中。通過select()方法查詢出銷售日期（date）和季度（quarter）字段。統(tǒng)計(jì)各季度房屋銷量和銷售額1．各季度房屋銷量統(tǒng)計(jì)通過groupBy()方法根據(jù)劃分的季度進(jìn)行分組，并統(tǒng)計(jì)每組的銷售數(shù)量。統(tǒng)計(jì)各季度房屋銷量和銷售額使用groupBy()方法根據(jù)季度進(jìn)行分組，統(tǒng)計(jì)每個(gè)季度的房屋銷售總額。統(tǒng)計(jì)各季度房屋銷量和銷售額2．各季度房屋銷售額統(tǒng)計(jì)使用groupBy()方法根據(jù)housing_rating字段進(jìn)行分組，并對(duì)每個(gè)分組下的房屋數(shù)量進(jìn)行統(tǒng)計(jì)，即可得出不同評(píng)分的房屋分布情況。探索分析房屋評(píng)分通過groupBy()方法根據(jù)housing