大數(shù)據(jù)導論思維第12章SPARKSQL課件

大數(shù)據(jù)導論第十二章大數(shù)據(jù)導論第十二章1CONTENTS目錄PART01SPARKSQL簡介PART02SPARKSQL執(zhí)行流程PART03基礎(chǔ)數(shù)據(jù)模型DATAFRAMEPART04使用SparkSQL的方式PART05SPARKSQL數(shù)據(jù)源PART06SPARKSQLCLI介紹PART07在Pyspark中使用SparkSQLPART08在Java中連接SparkSQLPART09習題CONTENTS目錄PART01SPARKSQL簡介2PART01SparkSQL簡介SparkSQL是一個用來處理結(jié)構(gòu)化數(shù)據(jù)的Spark組件，為Spark提供了查詢結(jié)構(gòu)化數(shù)據(jù)的能力。SparkSQL可被視為一個分布式的SQL查詢引擎，可以實現(xiàn)對多種數(shù)據(jù)格式和數(shù)據(jù)源進行SQL操作，包括Parquet，Hive，MongoDB，JSON、HDFS、JDBC、S3和RDD等。PART01SparkSQL簡介SparkSQL3SparkSQL簡介SparkSQL介紹：SparkSQL是為了處理結(jié)構(gòu)化數(shù)據(jù)的一個Spark模塊。不同于SparkRDD的基本API，SparkSQL接口擁有更多關(guān)于數(shù)據(jù)結(jié)構(gòu)本身與執(zhí)行計劃等更多信息。在Spark內(nèi)部，SparkSQL可以利用這些信息更好地對操作進行優(yōu)化。SparkSQL提供了三種訪問接口：SQL，DataFrameAPI和DatasetAPI。當計算引擎被用來執(zhí)行一個計算時，有不同的API和語言種類可供選擇。這種統(tǒng)一性意味著開發(fā)人員可以來回輕松切換各種最熟悉的API來完成同一個計算工作。SparkSQL簡介SparkSQL介紹：4SparkSQL簡介SparkSQL具有如下特點數(shù)據(jù)兼容方面：能加載和查詢來自各種來源的數(shù)據(jù)。性能優(yōu)化方面：除了采取內(nèi)存列存儲、代碼生成等優(yōu)化技術(shù)外，還引進成本模型對查詢進行動態(tài)評估、獲取最佳物理計劃等；組件擴展方面：無論是SQL的語法解析器、分析器還是優(yōu)化器都可以重新定義，進行擴展。標準連接：SparkSQL包括具有行業(yè)標準JDBC和ODBC連接的服務器模式。SparkSQL簡介SparkSQL具有如下特點5SparkSQL簡介SparkSQL具有如下特點集成：無縫地將SQL查詢與Spark程序混合。SparkSQL允許將結(jié)構(gòu)化數(shù)據(jù)作為Spark中的分布式數(shù)據(jù)集（RDD）進行查詢，在Python，Scala和Java中集成了API。這種緊密的集成使得SQL查詢以及復雜的分析算法可以輕松地運行?？蓴U展性：對于交互式查詢和長查詢使用相同的引擎。SparkSQL利用RDD模型來支持查詢?nèi)蒎e，使其能夠擴展到大型作業(yè)，不需擔心為歷史數(shù)據(jù)使用不同的引擎。SparkSQL簡介SparkSQL具有如下特點6PART02SparkSQL執(zhí)行流程PART02SparkSQL執(zhí)行流程7SparkSQL執(zhí)行流程類似于關(guān)系型數(shù)據(jù)庫，SparkSQL語句也是由Projection（a1，a2，a3）、DataSource（tableA）、Filter（condition）三部分組成，分別對應SQL查詢過程中的Result、DataSource、Operation，也就是說SQL語句按Result-->DataSource-->Operation的次序來描述的。SparkSQL執(zhí)行流程類似于關(guān)系型數(shù)據(jù)庫，SparkS8SparkSQL執(zhí)行流程解析（Parse）對讀入的SQL語句進行解析，分辨出SQL語句中哪些詞是關(guān)鍵詞（如SELECT、FROM、WHERE），哪些是表達式、哪些是Projection、哪些是DataSource等，從而判斷SQL語句是否規(guī)范；綁定（Bind）將SQL語句和數(shù)據(jù)庫的數(shù)據(jù)字典（列、表、視圖等）進行綁定，如果相關(guān)的Projection、DataSource等都存在，則這個SQL語句是可以執(zhí)行的；SparkSQL執(zhí)行流程解析（Parse）9SparkSQL執(zhí)行流程優(yōu)化（Optimize）一般的數(shù)據(jù)庫會提供幾個執(zhí)行計劃，這些計劃一般都有運行統(tǒng)計數(shù)據(jù)，數(shù)據(jù)庫會在這些計劃中選擇一個最優(yōu)計劃；執(zhí)行（Execute）按Operation-->DataSource-->Result的次序來執(zhí)行計劃。在執(zhí)行過程有時候甚至不需要讀取物理表就可以返回結(jié)果，比如重新運行剛運行過的SQL語句，可能直接從數(shù)據(jù)庫的緩沖池中獲取返回結(jié)果。SparkSQL執(zhí)行流程優(yōu)化（Optimize）10PART03基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame是由“命名列”（類似關(guān)系表的字段定義）所組織起來的一個分布式數(shù)據(jù)集合，可以把它看成是一個關(guān)系型數(shù)據(jù)庫的表。PART03基礎(chǔ)數(shù)據(jù)模型DataFrameDataFra11基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame是SparkSQL的核心，它將數(shù)據(jù)保存為行構(gòu)成的集合，行對應列有相應的列名。DataFrame與RDD的主要區(qū)別在于，DataFrame帶有Schema元信息，即DataFrame所表示的二維表數(shù)據(jù)集的每一列都帶有名稱和類型。這使得SparkSQL可以掌握更多的結(jié)構(gòu)信息，從而能夠?qū)ataFrame背后的數(shù)據(jù)源以及作用于DataFrame之上的變換進行了針對性的優(yōu)化，最終達到大幅提升運行時效率的目標?；A(chǔ)數(shù)據(jù)模型DataFrameDataFrame是Spark12基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame與RDD的對比：基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame與RDD的對13PART04使用SparkSQL的方式PART04使用SparkSQL的方式14使用SparkSQL的方式使用SparkSQL，首先利用sqlContext從外部數(shù)據(jù)源加載數(shù)據(jù)為DataFrame；然后，利用DataFrame上豐富的API進行查詢、轉(zhuǎn)換；最后，將結(jié)果進行展現(xiàn)或存儲為各種外部數(shù)據(jù)形式。Spark

SQL

為Spark提供了查詢結(jié)構(gòu)化數(shù)據(jù)的能力，查詢時既可以使用SQL也可以使用DataFrame

API（RDD）。通過ThriftServer，Spark

SQL支持多語言編程包括Java、Scala、Python及R。使用SparkSQL的方式使用SparkSQL，首先利用15使用SparkSQL的方式使用SparkSQL的方式16使用SparkSQL的方式加載數(shù)據(jù)①.從Hive中的users表構(gòu)造DataFrame：users=sqlContext.table("users")②.加載S3上的JSON文件：logs=sqlContext.load("s3n://path/to/data.json","json")③.加載HDFS上的Parquet文件：clicks=sqlContext.load("hdfs://path/to/data.parquet","parquet")使用SparkSQL的方式加載數(shù)據(jù)①.從Hive中的us17使用SparkSQL的方式加載數(shù)據(jù)④.通過JDBC訪問MySQL：comments=sqlContext.jdbc("jdbc:mysql://localhost/comments","user")⑤.將普通RDD轉(zhuǎn)變?yōu)镈ataFrame：rdd=sparkContext.textFile(“article.txt”).flatMap(_.split("")).map((_,1)).reduceByKey(_+_)wordCounts=sqlContext.createDataFrame(rdd,["word","count"])使用SparkSQL的方式加載數(shù)據(jù)④.通過JDBC訪問M18使用SparkSQL的方式加載數(shù)據(jù)⑥.將本地數(shù)據(jù)容器轉(zhuǎn)變?yōu)镈ataFrame：data=[("Alice",21),("Bob",24)]people=sqlContext.createDataFrame(data,["name","age"])⑦.將Pandas

DataFrame轉(zhuǎn)變?yōu)镾park

DataFrame（Python

API特有功能）：sparkDF

=

sqlContext.createDataFrame(pandasDF）使用SparkSQL的方式加載數(shù)據(jù)⑥.將本地數(shù)據(jù)容器轉(zhuǎn)變19使用SparkSQL的方式使用DataFrame①.創(chuàng)建一個只包含"年輕"用戶的DataFrame：young=users.filter(users.age<21)②.也可以使用Pandas風格的語法：young=users[users.age<21]③.將所有人的年齡加1：young.select(,young.age+1)使用SparkSQL的方式使用DataFrame①.創(chuàng)建20使用SparkSQL的方式使用DataFrame④.統(tǒng)計年輕用戶中各性別人數(shù)：young.groupBy("gender").count()⑤.將所有年輕用戶與另一個名為logs的DataFrame聯(lián)接起來：young.join(logs,logs.userId==users.userId,"left_outer")使用SparkSQL的方式使用DataFrame④.統(tǒng)計21使用SparkSQL的方式保存結(jié)果①.追加至HDFS上的Parquet文件：young.save(path="hdfs://path/to/data.parquet",source="parquet",mode="append")②.覆寫S3上的JSON文件：young.save(path="s3n://path/to/data.json",source="json",mode="append")使用SparkSQL的方式保存結(jié)果①.追加至HDFS上的22使用SparkSQL的方式保存結(jié)果③.保存為SQL表：young.saveAsTable(tableName="young",source="parquet"mode="overwrite")④.轉(zhuǎn)換為PandasDataFrame（PythonAPI特有功能）：pandasDF=young.toPandas()⑤.以表格形式打印輸出：young.show()使用SparkSQL的方式保存結(jié)果③.保存為SQL表：23SparkSQL通過DataFrame接口使用多種數(shù)據(jù)源。應用程序可以直接使用關(guān)系型轉(zhuǎn)換對DataFrame進行操作，也可以用來創(chuàng)建臨時表。把DataFrame注冊為臨時表后就可以使用SQL對其數(shù)據(jù)進行查詢PART05SparkSQL數(shù)據(jù)源SparkSQL通過DataFrame接口使用多種數(shù)據(jù)源。24通用Load方法SparkSQL使用數(shù)據(jù)源的最簡單模式就是對缺省數(shù)據(jù)源類型來進行各種操作。缺省數(shù)據(jù)源類型是parquet文件df=spark.read.load("examples/src/main/resources/users.parquet")df.select("name","favorite_color").write.save("namesAndFavColors.parquet")也可以通過load()函數(shù)的參數(shù)來說明數(shù)據(jù)源的類型：df=spark.read.load("examples/src/main/resources/people.json",format="json")df.select("name","age").write.save("namesAndAges.parquet",format="parquet")通用Load方法SparkSQL使用數(shù)據(jù)源的最簡單模式就是25通用Load方法也可以不把文件上載到DataFrame，然后再對其進行查詢操作，而是直接對文件進行SQL查詢：df=spark.sql("SELECT*FROMparquet.examples/src/main/resources/users.parquet")通用Load方法也可以不把文件上載到DataFrame，然后26通用Save方法Save方法Save()操作的可選參數(shù)SaveMode用來說明如何處理已有數(shù)據(jù)。需要提醒的是這些保存操作不是原子操作，如果選擇的是Overwrite模式，Save()操作會首先刪除已有的數(shù)據(jù)，然后才寫入新的數(shù)據(jù)。SaveMode的可選值：Scala/Java任何語言解釋SaveMode.ErrorIfExists"error"（缺省值）如果有數(shù)據(jù)存在，就拋異常。SaveMode.Append"append"如果有數(shù)據(jù)/表存在，就添加新數(shù)據(jù)到老數(shù)據(jù)后面。SaveMode.Overwrite"overwrite"如果有數(shù)據(jù)/表存在，就用新數(shù)據(jù)覆蓋老數(shù)據(jù)。SaveMode.Ignore"ignore"如果有數(shù)據(jù)存在，就不寫入新數(shù)據(jù)。通用Save方法Save方法Save()操作的可選參數(shù)Sav27通用Save方法基于文件的數(shù)據(jù)源，可以用path可選參數(shù)定義一個數(shù)據(jù)庫表的路徑，比如：df.write.option(“path”,“/some/path”).saveAsTable(“t”)即使數(shù)據(jù)庫表刪除后，該數(shù)據(jù)庫表路徑和數(shù)據(jù)也不會被刪除。如果程序不指定數(shù)據(jù)庫表路徑，Spark會把數(shù)據(jù)寫到一個缺省的數(shù)據(jù)庫表路徑下。但是在這種情況下，當數(shù)據(jù)庫被刪除的時候，缺省數(shù)據(jù)庫表的路徑將會被刪除。通用Save方法基于文件的數(shù)據(jù)源，可以用path可選參數(shù)定義28通用Save方法對基于文件的數(shù)據(jù)源還可以對其輸出進行分桶、排序和分割，分桶和排序只能用于當輸出方式為持久數(shù)據(jù)庫表的時候。df.write.bucketBy(42,"name").sortBy("age").saveAsTable("people_bucketed")當進行分割時，可以使用DatasetAPI接口的save()和saveAsTable()方法：df.write.partitionBy("favor_color").format("parquet").save("namesPartByColor.parquet")通用Save方法對基于文件的數(shù)據(jù)源還可以對其輸出進行分桶、排29通用Save方法還可以對同一張表同時進行分割和分桶：df=spark.read.parquet("examples/src/main/resources/users.parquet")(df.write.partitionBy("favorite_color").bucketBy(42,"name").saveAsTable("people_partitioned_bucketed"))通用Save方法還可以對同一張表同時進行分割和分桶：df=30Parquet文件數(shù)據(jù)源Parquet文件數(shù)據(jù)源SparkSQL支持對Parquet文件的讀寫操作，并自動保持源數(shù)據(jù)的模式。為了兼容性，在寫Parquet文件的時候，所有的列會自動轉(zhuǎn)換為可為空的列。把DataFrame保存為Parquet文件，格式信息會自動保留。peopleDF=spark.read.json("examples/src/main/resources/people.json")df.write.partitionBy("favor_color").format("parquet").save("namesPartByColor.parquet")Parquet文件數(shù)據(jù)源Parquet文件數(shù)據(jù)源Spark31讀取上面存儲的Parquet文件為DataFrameParquet文件也可以用來創(chuàng)建臨時表供SQL查詢使用parquetFile=spark.read.parquet("people.parquet")parquetFile.createOrReplaceTempView("parquetFile")teenagers=spark.sql("SELECTnameFROMparquetFileWHEREage>=13ANDage<=19")teenagers.show()Parquet文件數(shù)據(jù)源讀取上面存儲的Parquet文件為DataFrameParq32JSONDataSets數(shù)據(jù)源JSONDataSets數(shù)據(jù)源SparkSQL可以自動根據(jù)JSONDataSet的格式把其上載為DataFrame。用路徑指定JSONdataset；路徑下可以是一個文件，也可以是多個文件：sc=spark.sparkContextpath="examples/src/main/resources/people.json"peopleDF=spark.read.json(path)使用的結(jié)構(gòu)可以調(diào)用printSchema()方法打?。簆eopleDF.printSchema()JSONDataSets數(shù)據(jù)源JSONDataSets33利用DataFrame創(chuàng)建一個臨時表：使用Spark的sql方法進行SQL查詢：peopleDF.createOrReplaceTempView("people")teenagerNamesDF=spark.sql("SELECTnameFROMpeopleWHEREageBETWEEN13AND19")teenagerNamesDF.show()JSONDataSets數(shù)據(jù)源利用DataFrame創(chuàng)建一個臨時表：使用Spark的sql34JSONdataset的DataFrame也可以是RDD[String]格式，每個JSON對象為一個string：jsonStrings=['{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}']otherPeopleRDD=sc.parallelize(jsonStrings)otherPeople=spark.read.json(otherPeopleRDD)otherPeople.show()JSONDataSets數(shù)據(jù)源JSONdataset的DataFrame也可以是RDD[35Hive表數(shù)據(jù)源Hive表數(shù)據(jù)源SparkSQL支持對Hive中的數(shù)據(jù)進行讀寫。首先創(chuàng)建一個支持Hive的SparkSession對象，包括與Hivemetastore的連接，支持Hive的序列化和反序列化操作，支持用戶定義的Hive操作等。warehouse_location=abspath('spark-warehouse')spark=SparkSession.builder\.appName("PythonSparkSQLHiveintegrationexample")\.config("spark.sql.warehouse.dir",warehouse_location)\.enableHiveSupport().getOrCreate()warehouse_location指定數(shù)據(jù)庫和表的缺省位置：Hive表數(shù)據(jù)源Hive表數(shù)據(jù)源SparkSQL支持對Hi36Hive表數(shù)據(jù)源spark.sql("CREATETABLEIFNOTEXISTSsrc(keyINT,valueSTRING)USINGhive")spark.sql("LOADDATALOCALINPATH'examples/src/main/resources/kv1.txt'INTOTABLEsrc")基于新創(chuàng)建的SparkSession創(chuàng)建表和上載數(shù)據(jù)到表中：spark.sql("SELECT*FROMsrc").show()spark.sql("SELECTCOUNT(*)FROMsrc").show()使用HiveQL進行查詢：Hive表數(shù)據(jù)源spark.sql("CREATETABL37Hive表數(shù)據(jù)源sqlDF=spark.sql("SELECTkey,valueFROMsrcWHEREkey<10ORDERBYkey")SQL查詢的結(jié)果也是DataFrames，可以對結(jié)果進行所有DataFrame的操作：stringsDS=sqlDF.rdd.map(lambdarow:"Key:%d,Value:%s"%(row.key,row.value))forrecordinstringsDS.collect():print(record)DataFrames中的元素的類型是Row，所以可以按次序訪問列：Hive表數(shù)據(jù)源sqlDF=spark.sql("SEL38Hive表數(shù)據(jù)源Record=Row("key","value")recordsDF=spark.createDataFrame([Record(i,"val_"+str(i))foriinrange(1,101)])recordsDF.createOrReplaceTempView("records")也可以在同一個SparkSession內(nèi)用DataFrames來創(chuàng)建臨時表：Hive表數(shù)據(jù)源Record=Row("key","v39Hive表數(shù)據(jù)源創(chuàng)建Hive表的時候需要說明表通過哪種方式讀寫文件系統(tǒng)，以及如何序列化和反序列化數(shù)據(jù)。屬性名解釋fileFormat文件格式定義了存儲的格式，包括serde，inputformat和outputformat。目前支持6種文件格式：sequencefile，rcfile，orc，parquet，textfile和avro。inputFormat，outputFormat必須按對出現(xiàn)，分別定義存儲的inputformat和outputformat。不能與fileFormat同時使用。serde說明serde類的名字。當文件格式本身沒有serde時候使用。Hive表數(shù)據(jù)源創(chuàng)建Hive表的時候需要說明表通過哪種方式讀40SparkCLI指使用命令界面直接輸入SQL命令，然后發(fā)送到Spark集群進行執(zhí)行，在界面中顯示運行過程和最終的結(jié)果。PART06SparkSQLCLI介紹SparkCLI指使用命令界面直接輸入SQL命令，然后發(fā)送41SparkSQLCLI介紹SparkCLI指的是使用命令界面直接輸入SQL命令，然后發(fā)送到Spark集群進行執(zhí)行，在界面中顯示運行過程和最終的結(jié)果。SparkSQL已經(jīng)集成在SparkShell中，因此，只要啟動SparkShell，就可以使用SparkSQL的Shell交互接口：bin/spark-shell--masterspark://hdp-node-01:7077或者，可以啟動SparkSQL界面，使用起來更方便：bin/spark-sql--masterspark://hdp-node-01:7077SparkSQLCLI介紹SparkCLI指的是使用命42SparkSQLCLI介紹SparkSQL所有功能的入口是SQLContext類及其子類。為了創(chuàng)建一個基本的SQLContext，需要一個SparkContext。scala>valsqlContext=neworg.apache.spark.sql.SQLContext(sc)sqlContext:org.apache.spark.sql.SQLContext=org.apache.spark.sql.SQLContext@1943a343scala>importsqlContext.implicits._importsqlContext.implicits._SQLContextSparkSQLCLI介紹SparkSQL所有功能的入43SparkSQLCLI介紹下面的操作基于一個簡單的數(shù)據(jù)文件people.json，文件的內(nèi)容如下：{"name":"Michael"}{"name":"Andy","age":30}{"name":"Justin","age":19}數(shù)據(jù)文件下面語句從本地文件people.json讀取數(shù)據(jù)創(chuàng)建DataFrame：{valdf=sqlContext.read.json("file:///data/people.json")………df:org.apache.spark.sql.DataFrame=[age:bigint,name:string]創(chuàng)建DataFramesSparkSQLCLI介紹下面的操作基于一個簡單的數(shù)據(jù)文44Pyspark是針對Spark的PythonAPI。Spark使用py4j來實現(xiàn)Python與Java的互操作，從而實現(xiàn)使用Python編寫Spark程序。Spark也同樣提供了Pyspark，一個Spark的PythonShell，可以以交互的方式使用Python編寫Spark程序。PART07在Pyspark中使用SparkSQLPyspark是針對Spark的PythonAPI。Spa45在Pyspark中使用SparkSQL在終端上啟動Python

Spark

Shell：./bin/pyspark使用JSON文件作為數(shù)據(jù)源，創(chuàng)建JSON文件/home/sparksql/courses.json，并輸入下面的內(nèi)容：實例描述{"name":"Linux","type":"basic","length":10}{"name":"TCPIP","type":"project","length":15}{"name":"Python","type":"project","length":8}{"name":"GO","type":"basic","length":2}{"name":"Ruby","type":"basic","length":5}在Pyspark中使用SparkSQL在終端上啟動Pyth46在Pyspark中使用SparkSQL首先使用SQLContext模塊，其作用是提供SparkSQL處理的功能。在PysparkShell中逐步輸入下面步驟的內(nèi)容：引入pyspark.sql中的SQLContext：frompyspark.sqlimportSQLContext創(chuàng)建SQLContext對象使用pyspark的SparkContext對象，創(chuàng)建SQLContext對象：sqlContext=SQLContext(sc)在Pyspark中使用SparkSQL首先使用SQLCon47在Pyspark中使用SparkSQLDataFrame對象可以由RDD創(chuàng)建，也可以從Hive表或JSON文件等數(shù)據(jù)源創(chuàng)建。創(chuàng)建DataFrame，指明來源自JSON文件：df=sqlContext.read.json("/home/shiyanlou/courses.json")創(chuàng)建DataFrame對象在Pyspark中使用SparkSQLDataFrame對48在Pyspark中使用SparkSQL首先打印當前DataFrame里的內(nèi)容和數(shù)據(jù)表的格式：df.select("name").show()#展示了所有的課程名df.select("name","length").show()#展示了所有的課程名及課程長度對DataFrame進行操作show()函數(shù)將打印出JSON文件中存儲的數(shù)據(jù)表；使用printSchema()函數(shù)打印數(shù)據(jù)表的格式。然后對DataFrame的數(shù)據(jù)進行各種操作：df.show()df.printSchema()在Pyspark中使用SparkSQL首先打印當前Data49在Pyspark中使用SparkSQLdf.filter(df['type']=='basic').select('name','type').show()#展示了課程類型為基礎(chǔ)課（basic）的課程名和課程類型df.groupBy("type").count().show()#計算所有基礎(chǔ)課和項目課的數(shù)量。首先需要將DataFrame注冊為Table才可以在該表上執(zhí)行SQL語句：df.registerTempTable('courses')coursesRDD=sqlContext.sql("SELECTnameFROMcoursesWHERElength>=5andlength<=10")names=coursesRDD.rdd.map(lambdap:"Name:"+)fornameinnames.collect():printname執(zhí)行SQL語句在Pyspark中使用SparkSQLdf.filter(50在Pyspark中使用SparkSQLParquet是SparkSQL讀取的默認數(shù)據(jù)文件格式，把從JSON中讀取的DataFrame保存為Parquet格式，只保存課程名稱和長度兩項數(shù)據(jù)：df.select("name","length").write.save("/tmp/courses.parquet",format="parquet")保存DataFrame為其他格式將創(chuàng)建hdfs://master:9000/tmp/courses.parquet文件夾并存入課程名稱和長度數(shù)據(jù)。在Pyspark中使用SparkSQLParquet是Sp51SparkSQL實現(xiàn)了ThriftJDBC/ODBCserver，所以Java程序可以通過JDBC遠程連接SparkSQL發(fā)送SQL語句并執(zhí)行。PART08在Java中連接SparkSQLSparkSQL實現(xiàn)了ThriftJDBC/ODBCs52在Java中連接SparkSQL首先將${HIVE_HOME}/conf/hive-site.xml拷貝到${SPARK_HOME}/conf目錄下。另外，因為Hive元數(shù)據(jù)信息存儲在MySQL中，所以Spark在訪問這些元數(shù)據(jù)信息時需要MySQL連接驅(qū)動的支持。添加驅(qū)動的方式有三種：在${SPARK_HOME}/conf目錄下的spark-defaults.conf中添加：spark.jars/opt/lib2/mysql-connector-java-5.1.26-bin.jar

；可以實現(xiàn)添加多個依賴jar比較方便：spark.driver.extraClassPath/opt/lib2/mysql-connector-java-5.1.26-bin.jar；設(shè)置配置在Java中連接SparkSQL首先將${HIVE_HOM53在運行時添加--jars

/opt/lib2/mysql-connector-java-5.1.26-bin.jar做完上面的準備工作后，SparkSQL和Hive就繼承在一起了，SparkSQL可以讀取Hive中的數(shù)據(jù)。設(shè)置配置啟動Thrift在Spark根目錄下執(zhí)行：./sbin/start-thriftserver.sh開啟thrift服務器，它可以接受所有spark-submit的參數(shù)，并且還可以接受--hiveconf參數(shù)。不添加任何參數(shù)表示以local方式運行，默認的監(jiān)聽端口為10000在Java中連接SparkSQL在運行時添加--jars

/opt/lib2/mysql-54添加依賴打開Eclipse用JDBC連接HiveServer2。新建一個Maven項目，在pom.xml添加以下依賴：<dependency>

<groupId>org.apache.hive</groupId>

<artifactId>hive-jdbc</artifactId>

<version>1.2.1</version>

</dependency>

<dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-common</artifactId>

<version>2.4.1</version>

</dependency>

在Java中連接SparkSQL添加依賴打開Eclipse用JDBC連接HiveServe55添加依賴<dependency>

<groupId>jdk.tools</groupId>

<artifactId>jdk.tools</artifactId>

<version>1.6</version>

<scope>system</scope>

<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>

</dependency>

在Java中連接SparkSQL添加依賴<dependency>

在Java中連接Spar56JDBC連接HiveServer2的相關(guān)參數(shù)：驅(qū)動：org.apache.hive.jdbc.HiveDriverurl：jdbc:hive2://31:10000/default用戶名：hadoop(啟動thriftserver的linux用戶名)密碼：“”（默認密碼為空）JDBC連接參數(shù)在Java中連接SparkSQLJDBC連接HiveServer2的相關(guān)參數(shù)：JDBC連接57import

java.sql.Connection;

import

java.sql.DriverManager;

import

java.sql.ResultSet;

import

java.sql.SQLException;

import

java.sql.Statement;

public

class

Test1

{

public

static

void

main(String[]

args)

throws

SQLException

{

String

url

=

"jdbc:hive2://31:10000/default";

try

{

Class.forName("org.apache.hive.jdbc.HiveDriver");

}

catch

(ClassNotFoundException

e)

{

e.printStackTrace();

}

Connection

conn

=

DriverManager.getConnection(url,"hadoop","");

Statement

stmt

=

conn.createStatement();

String

sql

=

"SELECT

*

FROM

doc1

limit

10";

System.out.println("Running"+sql);

ResultSet

res

=

stmt.executeQuery(sql);

while(res.next()){

System.out.println("id:

"+res.getInt(1)+"\ttype:

"+res.getString(2)+"\tauthors:

"+res.getString(3)+"\ttitle:

"+res.getString(4)+"\tyear:"

+

res.getInt(5));

}

}

}

JDBC連接參數(shù)在Java中連接SparkSQLimport

java.sql.Connection;

J58PART09作業(yè)

PART09作業(yè)59作業(yè)作業(yè)：什么是SparkSQL？其主要目的是什么？SparkSQL的執(zhí)行流程有哪幾個步驟？在SparkSQL中，什么是DataFrame？使用DataFrame的優(yōu)勢是什么？DataFrame與RDD的主要區(qū)別是什么？使用SparkSQL的方式有哪幾種？使用SparkSQL的步驟是什么？常用的SparkSQL的數(shù)據(jù)源有哪些？Parquet文件格式是什么？它的主要特點是什么？為了使Java程序可以通過JDBC遠程連接SparkSQL，需要做哪些準備工作？連接數(shù)據(jù)庫的語句是什么？有哪些參數(shù)？作業(yè)作業(yè)：什么是SparkSQL？其主要目的是什么？60作業(yè)作業(yè)：請按下述要求寫出相應的SparkSQL語句：從一個本地JSON文件創(chuàng)建DataFrame；打印DataFrame元數(shù)據(jù)；按照列屬性過濾DataFrame的數(shù)據(jù)；返回某列滿足條件的數(shù)據(jù)；把DataFrame注冊成數(shù)據(jù)庫表。作業(yè)作業(yè)：請按下述要求寫出相應的SparkSQL語句：61謝謝FORYOURLISTENINGHandgeCO.LTD.2016.12.09謝謝FORYOURLISTENINGHandgeCO.62

大數(shù)據(jù)導論第十二章大數(shù)據(jù)導論第十二章63CONTENTS目錄PART01SPARKSQL簡介PART02SPARKSQL執(zhí)行流程PART03基礎(chǔ)數(shù)據(jù)模型DATAFRAMEPART04使用SparkSQL的方式PART05SPARKSQL數(shù)據(jù)源PART06SPARKSQLCLI介紹PART07在Pyspark中使用SparkSQLPART08在Java中連接SparkSQLPART09習題CONTENTS目錄PART01SPARKSQL簡介64PART01SparkSQL簡介SparkSQL是一個用來處理結(jié)構(gòu)化數(shù)據(jù)的Spark組件，為Spark提供了查詢結(jié)構(gòu)化數(shù)據(jù)的能力。SparkSQL可被視為一個分布式的SQL查詢引擎，可以實現(xiàn)對多種數(shù)據(jù)格式和數(shù)據(jù)源進行SQL操作，包括Parquet，Hive，MongoDB，JSON、HDFS、JDBC、S3和RDD等。PART01SparkSQL簡介SparkSQL65SparkSQL簡介SparkSQL介紹：SparkSQL是為了處理結(jié)構(gòu)化數(shù)據(jù)的一個Spark模塊。不同于SparkRDD的基本API，SparkSQL接口擁有更多關(guān)于數(shù)據(jù)結(jié)構(gòu)本身與執(zhí)行計劃等更多信息。在Spark內(nèi)部，SparkSQL可以利用這些信息更好地對操作進行優(yōu)化。SparkSQL提供了三種訪問接口：SQL，DataFrameAPI和DatasetAPI。當計算引擎被用來執(zhí)行一個計算時，有不同的API和語言種類可供選擇。這種統(tǒng)一性意味著開發(fā)人員可以來回輕松切換各種最熟悉的API來完成同一個計算工作。SparkSQL簡介SparkSQL介紹：66SparkSQL簡介SparkSQL具有如下特點數(shù)據(jù)兼容方面：能加載和查詢來自各種來源的數(shù)據(jù)。性能優(yōu)化方面：除了采取內(nèi)存列存儲、代碼生成等優(yōu)化技術(shù)外，還引進成本模型對查詢進行動態(tài)評估、獲取最佳物理計劃等；組件擴展方面：無論是SQL的語法解析器、分析器還是優(yōu)化器都可以重新定義，進行擴展。標準連接：SparkSQL包括具有行業(yè)標準JDBC和ODBC連接的服務器模式。SparkSQL簡介SparkSQL具有如下特點67SparkSQL簡介SparkSQL具有如下特點集成：無縫地將SQL查詢與Spark程序混合。SparkSQL允許將結(jié)構(gòu)化數(shù)據(jù)作為Spark中的分布式數(shù)據(jù)集（RDD）進行查詢，在Python，Scala和Java中集成了API。這種緊密的集成使得SQL查詢以及復雜的分析算法可以輕松地運行?？蓴U展性：對于交互式查詢和長查詢使用相同的引擎。SparkSQL利用RDD模型來支持查詢?nèi)蒎e，使其能夠擴展到大型作業(yè)，不需擔心為歷史數(shù)據(jù)使用不同的引擎。SparkSQL簡介SparkSQL具有如下特點68PART02SparkSQL執(zhí)行流程PART02SparkSQL執(zhí)行流程69SparkSQL執(zhí)行流程類似于關(guān)系型數(shù)據(jù)庫，SparkSQL語句也是由Projection（a1，a2，a3）、DataSource（tableA）、Filter（condition）三部分組成，分別對應SQL查詢過程中的Result、DataSource、Operation，也就是說SQL語句按Result-->DataSource-->Operation的次序來描述的。SparkSQL執(zhí)行流程類似于關(guān)系型數(shù)據(jù)庫，SparkS70SparkSQL執(zhí)行流程解析（Parse）對讀入的SQL語句進行解析，分辨出SQL語句中哪些詞是關(guān)鍵詞（如SELECT、FROM、WHERE），哪些是表達式、哪些是Projection、哪些是DataSource等，從而判斷SQL語句是否規(guī)范；綁定（Bind）將SQL語句和數(shù)據(jù)庫的數(shù)據(jù)字典（列、表、視圖等）進行綁定，如果相關(guān)的Projection、DataSource等都存在，則這個SQL語句是可以執(zhí)行的；SparkSQL執(zhí)行流程解析（Parse）71SparkSQL執(zhí)行流程優(yōu)化（Optimize）一般的數(shù)據(jù)庫會提供幾個執(zhí)行計劃，這些計劃一般都有運行統(tǒng)計數(shù)據(jù)，數(shù)據(jù)庫會在這些計劃中選擇一個最優(yōu)計劃；執(zhí)行（Execute）按Operation-->DataSource-->Result的次序來執(zhí)行計劃。在執(zhí)行過程有時候甚至不需要讀取物理表就可以返回結(jié)果，比如重新運行剛運行過的SQL語句，可能直接從數(shù)據(jù)庫的緩沖池中獲取返回結(jié)果。SparkSQL執(zhí)行流程優(yōu)化（Optimize）72PART03基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame是由“命名列”（類似關(guān)系表的字段定義）所組織起來的一個分布式數(shù)據(jù)集合，可以把它看成是一個關(guān)系型數(shù)據(jù)庫的表。PART03基礎(chǔ)數(shù)據(jù)模型DataFrameDataFra73基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame是SparkSQL的核心，它將數(shù)據(jù)保存為行構(gòu)成的集合，行對應列有相應的列名。DataFrame與RDD的主要區(qū)別在于，DataFrame帶有Schema元信息，即DataFrame所表示的二維表數(shù)據(jù)集的每一列都帶有名稱和類型。這使得SparkSQL可以掌握更多的結(jié)構(gòu)信息，從而能夠?qū)ataFrame背后的數(shù)據(jù)源以及作用于DataFrame之上的變換進行了針對性的優(yōu)化，最終達到大幅提升運行時效率的目標?；A(chǔ)數(shù)據(jù)模型DataFrameDataFrame是Spark74基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame與RDD的對比：基礎(chǔ)數(shù)據(jù)模型DataFrameDataFrame與RDD的對75PART04使用SparkSQL的方式PART04使用SparkSQL的方式76使用SparkSQL的方式使用SparkSQL，首先利用sqlContext從外部數(shù)據(jù)源加載數(shù)據(jù)為DataFrame；然后，利用DataFrame上豐富的API進行查詢、轉(zhuǎn)換；最后，將結(jié)果進行展現(xiàn)或存儲為各種外部數(shù)據(jù)形式。Spark

SQL

為Spark提供了查詢結(jié)構(gòu)化數(shù)據(jù)的能力，查詢時既可以使用SQL也可以使用DataFrame

API（RDD）。通過ThriftServer，Spark

SQL支持多語言編程包括Java、Scala、Python及R。使用SparkSQL的方式使用SparkSQL，首先利用77使用SparkSQL的方式使用SparkSQL的方式78使用SparkSQL的方式加載數(shù)據(jù)①.從Hive中的users表構(gòu)造DataFrame：users=sqlContext.table("users")②.加載S3上的JSON文件：logs=sqlContext.load("s3n://path/to/data.json","json")③.加載HDFS上的Parquet文件：clicks=sqlContext.load("hdfs://path/to/data.parquet","parquet")使用SparkSQL的方式加載數(shù)據(jù)①.從Hive中的us79使用SparkSQL的方式加載數(shù)據(jù)④.通過JDBC訪問MySQL：comments=sqlContext.jdbc("jdbc:mysql://localhost/comments","user")⑤.將普通RDD轉(zhuǎn)變?yōu)镈ataFrame：rdd=sparkContext.textFile(“article.txt”).flatMap(_.split("")).map((_,1)).reduceByKey(_+_)wordCounts=sqlContext.createDataFrame(rdd,["word","count"])使用SparkSQL的方式加載數(shù)據(jù)④.通過JDBC訪問M80使用SparkSQL的方式加載數(shù)據(jù)⑥.將本地數(shù)據(jù)容器轉(zhuǎn)變?yōu)镈ataFrame：data=[("Alice",21),("Bob",24)]people=sqlContext.createDataFrame(data,["name","age"])⑦.將Pandas

DataFrame轉(zhuǎn)變?yōu)镾park

DataFrame（Python

API特有功能）：sparkDF

=

sqlContext.createDataFrame(pandasDF）使用SparkSQL的方式加載數(shù)據(jù)⑥.將本地數(shù)據(jù)容器轉(zhuǎn)變81使用SparkSQL的方式使用DataFrame①.創(chuàng)建一個只包含"年輕"用戶的DataFrame：young=users.filter(users.age<21)②.也可以使用Pandas風格的語法：young=users[users.age<21]③.將所有人的年齡加1：young.select(,young.age+1)使用SparkSQL的方式使用DataFrame①.創(chuàng)建82使用SparkSQL的方式使用DataFrame④.統(tǒng)計年輕用戶中各性別人數(shù)：young.groupBy("gender").count()⑤.將所有年輕用戶與另一個名為logs的DataFrame聯(lián)接起來：young.join(logs,logs.userId==users.userId,"left_outer")使用SparkSQL的方式使用DataFrame④.統(tǒng)計83使用SparkSQL的方式保存結(jié)果①.追加至HDFS上的Parquet文件：young.save(path="hdfs://path/to/data.parquet",source="parquet",mode="append")②.覆寫S3上的JSON文件：young.save(path="s3n://path/to/data.json",source="json",mode="append")使用SparkSQL的方式保存結(jié)果①.追加至HDFS上的84使用SparkSQL的方式保存結(jié)果③.保存為SQL表：young.saveAsTable(tableName="young",source="parquet"mode="overwrite")④.轉(zhuǎn)換為PandasDataFrame（PythonAPI特有功能）：pandasDF=young.toPandas()⑤.以表格形式打印輸出：young.show()使用SparkSQL的方式保存結(jié)果③.保存為SQL表：85SparkSQL通過DataFrame接口使用多種數(shù)據(jù)源。應用程序可以直接使用關(guān)系型轉(zhuǎn)換對DataFrame進行操作，也可以用來創(chuàng)建臨時表。把DataFrame注冊為臨時表后就可以使用SQL對其數(shù)據(jù)進行查詢PART05SparkSQL數(shù)據(jù)源SparkSQL通過DataFrame接口使用多種數(shù)據(jù)源。86通用Load方法SparkSQL使用數(shù)據(jù)源的最簡單模式就是對缺省數(shù)據(jù)源類型來進行各種操作。缺省數(shù)據(jù)源類型是parquet文件df=spark.read.load("examples/src/main/resources/users.parquet")df.select("name","favorite_color").write.save("namesAndFavColors.parquet")也可以通過load()函數(shù)的參數(shù)來說明數(shù)據(jù)源的類型：df=spark.read.load("examples/src/main/resources/people.json",format="json")df.select("name","age").write.save("namesAndAges.parquet",format="parquet")通用Load方法SparkSQL使用數(shù)據(jù)源的最簡單模式就是87通用Load方法也可以不把文件上載到DataFrame，然后再對其進行查詢操作，而是直接對文件進行SQL查詢：df=spark.sql("SELECT*FROMparquet.examples/src/main/resources/users.parquet")通用Load方法也可以不把文件上載到DataFrame，然后88通用Save方法Save方法Save()操作的可選參數(shù)SaveMode用來說明如何處理已有數(shù)據(jù)。需要提醒的是這些保存操作不是原子操作，如果選擇的是Overwrite模式，Save()操作會首先刪除已有的數(shù)據(jù)，然后才寫入新的數(shù)據(jù)。SaveMode的可選值：Scala/Java任何語言解釋SaveMode.ErrorIfExists"error"（缺省值）如果有數(shù)據(jù)存在，就拋異常。SaveMode.Append"append"如果有數(shù)據(jù)/表存在，就添加新數(shù)據(jù)到老數(shù)據(jù)后面。SaveMode.Overwrite"overwrite"如果有數(shù)據(jù)/表存在，就用新數(shù)據(jù)覆蓋老數(shù)據(jù)。SaveMode.Ignore"ignore"如果有數(shù)據(jù)存在，就不寫入新數(shù)據(jù)。通用Save方法Save方法Save()操作的可選參數(shù)Sav89通用Save方法基于文件的數(shù)據(jù)源，可以用path可選參數(shù)定義一個數(shù)據(jù)庫表的路徑，比如：df.write.option(“path”,“/some/path”).saveAsTable(“t”)即使數(shù)據(jù)庫表刪除后，該數(shù)據(jù)庫表路徑和數(shù)據(jù)也不會被刪除。如果程序不指定數(shù)據(jù)庫表路徑，Spark會把數(shù)據(jù)寫到一個缺省的數(shù)據(jù)庫表路徑下。但是在這種情況下，當數(shù)據(jù)庫被刪除的時候，缺省數(shù)據(jù)庫表的路徑將會被刪除。通用Save方法基于文件的數(shù)據(jù)源，可以用path可選參數(shù)定義90通用Save方法對基于文件的數(shù)據(jù)源還可以對其輸出進行分桶、排序和分割，分桶和排序只能用于當輸出方式為持久數(shù)據(jù)庫表的時候。df.write.bucketBy(42,"name").sortBy("age").saveAsTable("people_bucketed")當進行分割時，可以使用DatasetAPI接口的save()和saveAsTable()方法：df.write.partitionBy("favor_color").format("parquet").save("namesPartByColor.parquet")通用Save方法對基于文件的數(shù)據(jù)源還可以對其輸出進行分桶、排91通用Save方法還可以對同一張表同時進行分割和分桶：df=spark.read.parquet("examples/src/main/resources/users.parquet")(df.write.partitionBy("favorite_color").bucketBy(42,"name").saveAsTable("people_partitioned_bucketed"))通用Save方法還可以對同一張表同時進行分割和分桶：df=92Parquet文件數(shù)據(jù)源Parquet文件數(shù)據(jù)源SparkSQL支持對Parquet文件的讀寫操作，并自動保持源數(shù)據(jù)的模式。為了兼容性，在寫Parquet文件的時候，所有的列會自動轉(zhuǎn)換為可為空的列。把DataFrame保存為Parquet文件，格式信息會自動保留。peopleDF=spark.read.json("examples/src/main/resources/people.json")df.write.partitionBy("favor_color").format("parquet").save("namesPartByColor.parquet")Parquet文件數(shù)據(jù)源Parquet文件數(shù)據(jù)源Spark93讀取上面存儲的Parquet文件為DataFrameParquet文件也可以用來創(chuàng)建臨時表供SQL查詢使用parquetFile=spark.read.parquet("people.parquet")parquetFile.createOrReplaceTempView("parquetFile")teenagers=spark.sql("SELECTnameFROMparquetFileWHEREage>=13ANDage<=19")teenagers.show()Parquet文件數(shù)據(jù)源讀取上面存儲的Parquet文件為DataFrameParq94JSONDataSets數(shù)據(jù)源JSONDataSets數(shù)據(jù)源SparkSQL可以自動根據(jù)JSONDataSet的格式把其上載為DataFrame。用路徑指定JSONdataset；路徑下可以是一個文件，也可以是多個文件：sc=spark.sparkContextpath="examples/src/main/resources/people.json"peopleDF=spark.read.json(path)使用的結(jié)構(gòu)可以調(diào)用printSchema()方法打?。簆eopleDF.printSchema()JSONDataSets數(shù)據(jù)源JSONDataSets95利用DataFrame創(chuàng)建一個臨時表：使用Spark的sql方法進行SQL查詢：peopleDF.createOrReplaceTempView("people")teenagerNamesDF=spark.sql("SELECTnameFROMpeopleWHEREageBETWEEN13AND19")teenagerNamesDF.show()JSONDataSets數(shù)據(jù)源利用DataFrame創(chuàng)建一個臨時表：使用Spark的sql96JSONdataset的DataFrame也可以是RDD[String]格式，每個JSON對象為一個string：jsonStrings=['{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}']otherPeopleRDD=sc.parallelize(jsonStrings)otherPeople=spark.read.json(otherPeopleRDD)otherPeople.show()JSONDataSets數(shù)據(jù)源JSONdataset的DataFrame也可以是RDD[97Hive表數(shù)據(jù)源Hive表數(shù)據(jù)源SparkSQL支持對Hive中的數(shù)據(jù)進行讀寫。首先創(chuàng)建一個支持Hive的SparkSession對象，包括與Hivemetastore的連接，支持Hive的序列化和反序列化操作，支持用戶定義的Hive操作等。warehouse_location=abspath('spark-warehouse')spark=SparkSession.builder\.appName("PythonSparkSQLHiveintegrationexample")\.config("spark.sql.warehouse.dir",warehouse_location)\.enableHiveSupport().getOrCreate()warehouse_location指定數(shù)據(jù)庫和表的缺省位置：Hive表數(shù)據(jù)源Hive表數(shù)據(jù)源SparkSQL支持對Hi98Hive表數(shù)據(jù)源spark.sql("CREATETABLEIFNOTEXISTSsrc(keyINT,valueSTRING)USINGhive")spark.sql("LOADDATALOCALINPATH'examples/src/main/resources/kv1.txt'INTOTABLEsrc")基于新創(chuàng)建的SparkSession創(chuàng)建表和上載數(shù)據(jù)到表中：spark.sql("SELECT*FROMsrc").show()spark.sql("SELECTCOUNT(*)FROMsrc").show()使用HiveQL進行查詢：Hive表數(shù)據(jù)源spark.sql("CREATETABL99Hive表數(shù)據(jù)源sqlDF=spark.sql("SELECTkey,valueFROMsrcWHEREkey<10ORDERBYkey")SQL查詢的結(jié)果也是DataFrames，可以對結(jié)果進行所有DataFrame的操作：stringsDS=sqlDF.rdd.map(lambdaro