SparkSQL第6章目

錄01SparkSQL簡(jiǎn)介02結(jié)構(gòu)化數(shù)據(jù)DataFrame03DataFrame的創(chuàng)建和保存04DataFrame的常用操作05從RDD轉(zhuǎn)換得到DataFrame06使用SparkSQL讀寫數(shù)據(jù)庫(kù)07DataSetSparkSQL簡(jiǎn)介6.1SparkSQL簡(jiǎn)介從Shark說起0102SparkSQL設(shè)計(jì)03為什么推出SparkSQLSparkSQL特點(diǎn)04編程實(shí)例056.1.1從Shark說起ClientMetastoreJDBCDriverSQLParserQueryOptimizerPhysicalplanExecutionMapReduceHDFSCLIHive:SQL-on-Hadoop6.1.1從Shark說起輸入ParserSemanticAnalyzerLogicalPlanGeneratorLogicalOptimizerPhysicalPlanGeneratorPhysicalOptimizer將SQL轉(zhuǎn)換成抽象語法樹將抽象語法樹轉(zhuǎn)換成查詢塊將查詢塊轉(zhuǎn)換成邏輯查詢計(jì)劃重寫邏輯查詢計(jì)劃將邏輯查詢計(jì)劃轉(zhuǎn)成物理計(jì)劃選擇最佳的優(yōu)化查詢策略輸出6.1.1從Shark說起Shark即HiveonSpark，為了實(shí)現(xiàn)與Hive兼容，Shark在HiveQL方面重用了Hive中HiveQL的解析、邏輯執(zhí)行計(jì)劃翻譯、執(zhí)行計(jì)劃優(yōu)化等邏輯，可以近似認(rèn)為僅將物理執(zhí)行計(jì)劃從MapReduce作業(yè)替換成了Spark作業(yè)，通過Hive的HiveQL解析，把HiveQL翻譯成Spark上的RDD操作6.1.1從Shark說起Shark使SQL-on-Hadoop性能比Hive提高了10-100倍6.1.1從Shark說起1Shark導(dǎo)致的兩個(gè)問題???完全依賴執(zhí)行計(jì)劃優(yōu)化不方便添加新的優(yōu)化策略6.1.1從Shark說起Spark線程級(jí)并行MapReduce進(jìn)程級(jí)并行Shark導(dǎo)致的兩個(gè)問題2???6.1.1從Shark說起線程安全問題Spark兼容2???Shark導(dǎo)致的兩個(gè)問題6.1.1從Shark說起停止對(duì)Shark的開發(fā)2014年6月1日6.1.1從Shark說起AnewSQLenginedesignedFromground-upforSparkSparkSQLSharkDevelopmentending；TransitioningtoSparkSQLHelpexisitingHiveuseMigratetoSparkHiveonSparkHive又能夠支持MapReduce6.1.1從Shark說起SparkSQL作為Spark生態(tài)的一員不再受限于Hive，只是兼容Hive01HiveonSpark是一個(gè)Hive發(fā)展計(jì)劃，該計(jì)劃將Spark作為Hive的底層引擎之一，Hive將不再受限于一個(gè)引擎，可以采用Map-Reduce、Tez、Spark等引擎026.1.2SparkSQL架構(gòu)ClientMetastoreJDBCDriverSQLParserQueryOptimizerPhysicalplanExecutionMapReduceHDFSCLIHive:SQL-on-HadoopSparkSQL在Hive兼容層面僅依賴HiveQL解析、Hive元數(shù)據(jù)，從HQL被解析成抽象語法樹（AST）起，就全部由SparkSQL接管了。SparkSQL執(zhí)行計(jì)劃生成和優(yōu)化都由Catalyst（函數(shù)式關(guān)系查詢優(yōu)化框架）負(fù)責(zé)SparkSQL支持的數(shù)據(jù)格式和編程語言6.1.2SparkSQL架構(gòu)6.1.3為什么推出SparkSQL6.1.3為什么推出SparkSQL6.1.3為什么推出SparkSQL關(guān)系數(shù)據(jù)庫(kù)存儲(chǔ)一部分結(jié)構(gòu)化數(shù)據(jù)6.1.3為什么推出SparkSQL結(jié)構(gòu)化數(shù)據(jù)非結(jié)構(gòu)化數(shù)據(jù)大數(shù)據(jù)10%90%存儲(chǔ)在關(guān)系數(shù)據(jù)庫(kù)中與人類信息密切相關(guān)6.1.3為什么推出SparkSQL結(jié)構(gòu)化數(shù)據(jù)非結(jié)構(gòu)化數(shù)據(jù)半結(jié)構(gòu)化數(shù)據(jù)關(guān)系數(shù)據(jù)庫(kù)在大數(shù)據(jù)時(shí)代已經(jīng)不能滿足要求不能用SQL語句6.1.3為什么推出SparkSQL需要執(zhí)行高級(jí)分析機(jī)器學(xué)習(xí)算法決策樹數(shù)據(jù)分析6.1.3為什么推出SparkSQL傳統(tǒng)關(guān)系數(shù)據(jù)庫(kù)數(shù)據(jù)分析SQL機(jī)器學(xué)習(xí)算法分析6.1.3為什么推出SparkSQL圖像處理機(jī)器學(xué)習(xí)關(guān)系數(shù)據(jù)庫(kù)沒辦法處理6.1.3為什么推出SparkSQL關(guān)系查詢復(fù)雜分析算法能夠處理結(jié)構(gòu)化、半結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)6.1.3為什么推出SparkSQLSQL非結(jié)構(gòu)化數(shù)據(jù)半結(jié)構(gòu)化數(shù)據(jù)構(gòu)建DataFrame6.1.3為什么推出SparkSQLSQL非結(jié)構(gòu)化數(shù)據(jù)半結(jié)構(gòu)化數(shù)據(jù)構(gòu)建DataFrame文本類型進(jìn)行解析相關(guān)查詢DataFrame關(guān)系型表格6.1.3為什么推出SparkSQL文本類型進(jìn)行解析相關(guān)查詢關(guān)系型表格能夠融合非結(jié)構(gòu)化數(shù)據(jù)分析方便調(diào)用6.1.3為什么推出SparkSQL關(guān)系數(shù)據(jù)查詢復(fù)雜分析算法SQL6.1.3為什么推出SparkSQL機(jī)器學(xué)習(xí)算法的數(shù)據(jù)處理能力關(guān)系數(shù)據(jù)庫(kù)的結(jié)構(gòu)化數(shù)據(jù)管理能力融合SparkSQL6.1.4

SparkSQL的特點(diǎn)SparkSQL可將SQL查詢和Spark程序無縫集成允許使用SQL或熟悉的DataFrameAPI在Spark程序中查詢結(jié)構(gòu)化數(shù)據(jù)容易整合（集成）DataFrame和SQL提供了訪問各種數(shù)據(jù)源的方法，包括Hive、Avro、Parquet、ORC、JSON和JDBC統(tǒng)一的數(shù)據(jù)訪問方式parkSQL支持HiveQL語法以及HiveSerDes和UDF允許我們?cè)L問現(xiàn)有的Hive倉(cāng)庫(kù)兼容HiveSparkSQL支持JDBC或ODBC連接標(biāo)準(zhǔn)的數(shù)據(jù)庫(kù)連接6.1.5SparkSQL簡(jiǎn)單編程實(shí)例SparkSession接口SparkSession實(shí)現(xiàn)了SQLContext及HiveContext所有功能。此外，SparkSession也封裝了SparkConf、SparkContext和StreamingContextSparkSession支持從不同的數(shù)據(jù)源加載數(shù)據(jù)，以及把數(shù)據(jù)轉(zhuǎn)換成DataFrame，并且支持把DataFrame轉(zhuǎn)換成SQLContext自身的表，然后使用SQL語句來操作數(shù)據(jù)可以通過如下語句創(chuàng)建一個(gè)SparkSession對(duì)象importorg.apache.spark.sql.SparkSessionvalspark=SparkSession.builder().appName(“SparkSessionExample”).getOrCreate()6.1.5SparkSQL簡(jiǎn)單編程實(shí)例在Linux終端中，執(zhí)行如下命令創(chuàng)建一個(gè)目錄sparkapp作為應(yīng)用程序根目錄$cd~#進(jìn)入用戶主目錄$mkdir./sparkapp#創(chuàng)建應(yīng)用程序根目錄，如果已經(jīng)存在，則不用創(chuàng)建$mkdir-p./sparkapp/src/main/scala

#創(chuàng)建所需的目錄結(jié)構(gòu)6.1.5SparkSQL簡(jiǎn)單編程實(shí)例下面給出一個(gè)具體實(shí)例，介紹SparkSQL的編程方法6.1.5SparkSQL簡(jiǎn)單編程實(shí)例importorg.apache.spark.sql.SparkSessionobjectSimpleApp{defmain(args:Array[String]){vallogFile="file:///usr/local/spark/README.md"valspark=SparkSession.builder.appName("SimpleApplication").getOrCreate()vallogData=spark.read.textFile(logFile).cache()valnumAs=logData.filter(line=>line.contains("a")).count()valnumBs=logData.filter(line=>line.contains("b")).count()println(s"Lineswitha:$numAs,Lineswithb:$numBs")spark.stop()}}在“src/main/scala”目錄下創(chuàng)建一個(gè)代碼文件SimpleApp.scala，其內(nèi)容如下

6.1.5SparkSQL簡(jiǎn)單編程實(shí)例name:="SimpleProject"version:="1.0"scalaVersion:="2.12.15"libraryDependencies+="org.apache.spark"%%"spark-sql"%"3.2.0"

在sparkapp目錄下創(chuàng)建一個(gè)文件simple.sbt，并設(shè)置為如下內(nèi)容6.1.5SparkSQL簡(jiǎn)單編程實(shí)例$/usr/local/sbt/sbtpackage然后，使用spark-submit命令運(yùn)行程序，就可以得到執(zhí)行結(jié)果，具體如下：$cd~/sparkapp$/usr/local/spark/bin/spark-submit\>--class"SimpleApp"\>./target/scala-2.12/simple-project_2.12-1.0.jar2>&1|grep"Lineswitha:"

執(zhí)行如下代碼使用sbt工具對(duì)代碼進(jìn)行編譯打包在Linux終端中，執(zhí)行如下命令創(chuàng)建一個(gè)目錄sparkapp作為應(yīng)用程序根目錄Lineswitha:65,Lineswithb:33結(jié)構(gòu)化數(shù)據(jù)DataFrame6.2結(jié)構(gòu)化數(shù)據(jù)DataFrame6.2.1DataFrame概述6.2.2DataFrame的優(yōu)點(diǎn)6.2.1DataFrame概述DataFrameRDD(Person)DataFrame與RDD的區(qū)別PersonPersonPersonPersonPersonPersonNameAgeHeightStringIntDoubleStringIntDoubleStringIntDoubleStringIntDoubleStringIntDoubleStringIntDouble6.2.1DataFrame概述處理大規(guī)模結(jié)構(gòu)化數(shù)據(jù)SparkSQL

MySQL→DataFrame支持SQL查詢6.2.1DataFrame概述倉(cāng)庫(kù)保存洗漱用具6.2.1DataFrame概述張三傳統(tǒng)RDD方式6.2.1DataFrame概述張三張三打開傳統(tǒng)RDD方式6.2.1DataFrame概述DataFrame方式張三DataFrame方式6.2.1DataFrame概述張三張三RDD是分布式的Java對(duì)象的集合，但是，對(duì)象內(nèi)部結(jié)構(gòu)對(duì)于RDD而言卻是不可知的DataFrame是一種以RDD為基礎(chǔ)的分布式數(shù)據(jù)集，提供了詳細(xì)的結(jié)構(gòu)信息6.2.1DataFrame概述6.2.2

DataFrame的優(yōu)點(diǎn)風(fēng)格一致更優(yōu)的空間效率更好的性能易組合簡(jiǎn)潔表達(dá)能力強(qiáng)優(yōu)

點(diǎn)6.2.2

DataFrame的優(yōu)點(diǎn)("spark",2)("hadoop",6)("hadoop",4)("spark",6)scala>valbookRDD=sc.|parallelize(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6)))scala>valsaleRDD=bookRDD.map(x=>(x._1,(x._2,1))).|reduceByKey((x,y)=>(x._1+y._1,x._2+y._2)).|map(x=>(x._1,x._2._1/x._2._2)).collect()計(jì)算每個(gè)鍵對(duì)應(yīng)的平均值，即每種圖書的每天平均銷量。使用RDD編程時(shí)，語句如下

6.2.2

DataFrame的優(yōu)點(diǎn)scala>valbookDF=spark.|createDataFrame(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6))).|toDF("book","amount")scala>valavgDF=bookDF.groupBy("book").agg(avg("amount"))scala>avgDF.show()+------+-----------+|book|avg(amount)|+------+-----------+|spark|4.0||hadoop|5.0|+------+-----------+如果是使用DataFrameAPI來表達(dá)相同的查詢，就會(huì)簡(jiǎn)單很多提高了代碼的表達(dá)能力實(shí)現(xiàn)更高的執(zhí)行效率DataFrame的創(chuàng)建和保存6.3DataFrame的創(chuàng)建和保存6.3.2JSON6.3.4文本文件6.3.1Parquet6.3.3CSV010203046.3.1Parquetscala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/users.parquet"scala>valdf=spark.read.format("parquet").load(filePath)scala>df.show()+------+--------------+----------------+|name|favorite_color|favorite_numbers|+------+--------------+----------------+|Alyssa|null|[3,9,15,20]||Ben|red|[]|+------+--------------+----------------+

1.

從Parquet文件創(chuàng)建DataFrame6.3.1Parquet或者也可以使用如下方式讀取Parquet文件生成DataFramescala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/users.parquet"scala>valdf=spark.read.parquet(filePath)6.3.1Parquet

2.

將DataFrame保存為Parquet文件（在上面的代碼基礎(chǔ)上繼續(xù)執(zhí)行下面代碼）scala>df.write.format("parquet").mode("overwrite").option("compression","snappy").|save("file:///home/hadoop/otherusers")上面代碼執(zhí)行以后，在本地文件系統(tǒng)中的“/home/hadoop/”目錄下會(huì)生成一個(gè)名稱為“otherusers”的子目錄，該目錄下包含兩個(gè)文件，即_SUCCESS文件和像part-00000-XXXX.snappy.parquet這樣的文件，后者是使用snappy壓縮算法得到的壓縮文件。如果要再次讀取文件生成DataFrame，load()中可以直接使用目錄“file:///home/hadoop/otherusers”也可以使用文件“file:///home/hadoop/otherusers/part-00000-XXXX.snappy.parquet”scala>df.write.parquet("file:///home/hadoop/otherusers")6.3.2JSONscala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.json"scala>valdf=spark.read.format("json").load(filePath)scala>df.show()+----+-------+|age|name|+----+-------+|null|Michael||30|Andy||19|Justin|+----+-------+

1.

從JSON文件創(chuàng)建DataFrame：從JSON文件創(chuàng)建DataFrame的具體方法如下或者也可以使用如下方式讀取JSON文件生成DataFramescala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.json"scala>valdf=spark.read.json(filePath)6.3.2JSON

2.

將DataFrame保存為JSON文件（在上面的代碼基礎(chǔ)上繼續(xù)執(zhí)行下面代碼）scala>df.write.format("json").mode("overwrite").|save("file:///home/hadoop/otherpeople")上面代碼執(zhí)行以后，在本地文件系統(tǒng)中的“/home/hadoop/”目錄下會(huì)生成一個(gè)名稱為“otherpeople”的子目錄，該目錄下包含兩個(gè)文件，即_SUCCESS文件和像part-00000-XXXX.json這樣的文件。如果要再次讀取文件生成DataFrame，load()中可以直接使用目錄“file:///home/hadoop/otherpeople”，也可以使用文件“file:///home/hadoop/otherpeople/part-00000-XXXX.json”scala>df.write.json("file:///home/hadoop/otherpeople")6.3.2JSON6.3.3CSVscala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.csv"scala>valschema="nameSTRING,ageINT,jobSTRING"scala>valdf=spark.read.format("csv").schema(schema).option("header","true").|option("sep",";").load(filePath)scala>df.show()+-----+---+---------+|name|age|job|+-----+---+---------+|Jorge|30|Developer||Bob|32|Developer|+-----+---+---------+

1.

從CSV文件創(chuàng)建DataFrame：從CSV文件創(chuàng)建DataFrame的具體方法如下或者也可以使用如下方式讀取CSV文件生成DataFrame6.3.3CSVscala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.csv"scala>valschema="nameSTRING,ageINT,jobSTRING"scala>valdf=spark.read.schema(schema).option("header","true").|option("sep",";").csv(filePath)

2.

將DataFrame保存為CSV文件（在上面的代碼基礎(chǔ)上繼續(xù)執(zhí)行下面代碼）scala>df.write.format("csv").mode("overwrite").|save("file:///home/hadoop/anotherpeople")上面代碼執(zhí)行以后，在本地文件系統(tǒng)中的“/home/hadoop/”目錄下會(huì)生成一個(gè)名稱為“anotherpeople”的子目錄，該目錄下包含兩個(gè)文件，即_SUCCESS文件和像part-00000-XXXX.csv這樣的文件。如果要再次讀取文件生成DataFrame，load()中可以直接使用目錄“file:///home/hadoop/anotherpeople”，也可以使用文件“file:///home/hadoop/anotherpeople/part-00000-XXXX.csv”scala>df.write.csv("file:///home/hadoop/anotherpeople")6.3.3CSV6.3.4文本文件scala>valfilePath="file:///home/hadoop/word.txt"scala>valdf=spark.read.format("text").load(filePath)scala>df.show()+---------------+|value|+---------------+|hadoopisgood||sparkisbetter||sparkisfast|+---------------+

從一個(gè)文本文件創(chuàng)建DataFrame的方法如下或者也可以使用如下方式讀取文本文件生成DataFramescala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.txt"scala>valdf=spark.read.text(filePath)6.3.4文本文件

如果要把一個(gè)DataFrame保存成文本文件，則需要使用如下語句格式scala>valpeopleDF=spark.read.format("json").|load("file:///usr/local/spark/examples/src/main/resources/people.json")scala>peopleDF.rdd.saveAsTextFile("file:///home/hadoop/newpeople")上面代碼執(zhí)行以后，會(huì)在生成目錄"file:///home/hadoop/newpeople"，這個(gè)目錄下會(huì)包含兩個(gè)文件，part-00000和_SUCCESS，其中，part-00000文件中包含了具體數(shù)據(jù)scala>df.write.format("text").save("file:///home/hadoop/newpeople")6.3.4文本文件DataFrame的基本操作6.4DataFrame的基本操作DSL語法風(fēng)格SQL語法風(fēng)格6.4.1DSL語法風(fēng)格DSL:“領(lǐng)域?qū)Ｓ谜Z言”允許開發(fā)者通過調(diào)用方法對(duì)DataFrame內(nèi)部的數(shù)據(jù)進(jìn)行分析6.4.1DSL語法風(fēng)格show()withColumn()filter()groupBy()select()sort()printSchema()drop()6.4.1DSL語法風(fēng)格scala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.json"scala>valdf=spark.read.json(filePath)

首先從Spark自帶的樣例文件people.json創(chuàng)建一個(gè)名稱為df的DataFrame6.4.1DSL語法風(fēng)格.printSchema().show()6.4.1DSL語法風(fēng)格.select()6.4.1DSL語法風(fēng)格.filter().groupBy()6.4.1DSL語法風(fēng)格.sort()6.4.1DSL語法風(fēng)格.withColumn()6.4.1DSL語法風(fēng)格.drop()6.4.1DSL語法風(fēng)格其他常用操作6.4.2

SQL語法風(fēng)格SQL使用SQL語句操作DataFrameSQL函數(shù)6.4.2

SQL語法風(fēng)格熟練使用SQL語法的開發(fā)者，可直接使用SQL語句進(jìn)行數(shù)據(jù)操作。相比于DSL語法風(fēng)格，在執(zhí)行SQL語句之前，需通過DataFrame實(shí)例創(chuàng)建臨時(shí)視圖。創(chuàng)建臨時(shí)視圖的方法是調(diào)用DataFrame的createTempView或createOrReplaceTempView方法，二者的區(qū)別是，后者會(huì)進(jìn)行判斷01對(duì)于createOrReplaceTempView方法而言，如果在當(dāng)前會(huì)話中存在相同名稱的臨時(shí)視圖，則用新視圖替換原來的臨時(shí)視圖，如果在當(dāng)前會(huì)話中不存在相同名稱的臨時(shí)視圖，則創(chuàng)建臨時(shí)視圖。對(duì)于createTempView方法而言，如果在當(dāng)前會(huì)話中存在相同名稱的臨時(shí)視圖，則會(huì)直接報(bào)錯(cuò)02scala>valfilePath=|"file:///usr/local/spark/examples/src/main/resources/people.json"scala>valdf=spark.read.json(filePath)scala>df.show()+----+-------+|age|name|+----+-------+|null|Michael||30|Andy||19|Justin|+----+-------+

1.

使用SQL語句操作DataFrame6.4.2

SQL語法風(fēng)格scala>df.createTempView("people")scala>spark.sql("SELECT*FROMpeople").show()+----+-------+|age|name|+----+-------+|null|Michael||30|Andy||19|Justin|+----+-------+scala>spark.sql("SELECTnameFROMpeoplewhereage>20").show()+----+|name|+----+|Andy|+----+6.4.2

SQL語法風(fēng)格6.4.2

SQL語法風(fēng)格2.

SQL函數(shù)：提供了豐富的函數(shù)供用戶選擇一共200多個(gè)，基本涵蓋了大部分的日常應(yīng)用場(chǎng)景，包括轉(zhuǎn)換函數(shù)、數(shù)學(xué)函數(shù)、字符串函數(shù)、二進(jìn)制函數(shù)、日期時(shí)間函數(shù)、正則表達(dá)式函數(shù)、JSON函數(shù)、URL函數(shù)、聚合函數(shù)、窗口函數(shù)和集合函數(shù)等。當(dāng)Spark自帶的這些系統(tǒng)函數(shù)無法滿足用戶需求時(shí)，用戶還可以創(chuàng)建“用戶自定義函數(shù)”scala>importorg.apache.spark.sql.Rowscala>importorg.apache.spark.sql.types._scala>valschema=StructType(List(StructField("name",StringType,true),|StructField("age",IntegerType,true),|StructField("create_time",LongType,true)))scala>valjavaList=newjava.util.ArrayList[Row]()scala>javaList.add(Row("Xiaomei",21,System.currentTimeMillis()/1000))scala>javaList.add(Row("Xiaoming",22,System.currentTimeMillis()/1000))scala>javaList.add(Row("Xiaoxue",23,System.currentTimeMillis()/1000))scala>valdf=spark.createDataFrame(javaList,schema)

使用用戶自定義函數(shù)將用戶名轉(zhuǎn)化為大寫英文字母。具體實(shí)現(xiàn)代碼如下6.4.2

SQL語法風(fēng)格nameagecreate_timescala>df.show()+--------+---+-----------+|name|age|create_time|+--------+---+-----------+|Xiaomei|21|1644480595||Xiaoming|22|1644480607||Xiaoxue|23|1644480615|+--------+---+-----------+scala>df.createTempView("user_info")scala>spark.sql("SELECTname,age,from_unixtime(create_time,'yyyy-MM-ddHH:mm:ss')FROMuser_info").show()+--------+---+-----------------------------------------------+|name|age|from_unixtime(create_time,yyyy-MM-ddHH:mm:ss)|+--------+---+-----------------------------------------------+|Xiaomei|21|2022-02-1000:09:55||Xiaoming|22|2022-02-1000:10:07||Xiaoxue|23|2022-02-1000:10:15|+--------+---+-----------------------------------------------+6.4.2

SQL語法風(fēng)格scala>spark.udf.register("toUpperCaseUDF",(column:String)=>column.toUpperCase)scala>spark.sql("SELECTtoUpperCaseUDF(name),age,from_unixtime(create_time,'yyyy-MM-ddHH:mm:ss')FROMuser_info").show()+--------------------+---+-----------------------------------------------+|toUpperCaseUDF(name)|age|from_unixtime(create_time,yyyy-MM-ddHH:mm:ss)|+--------------------+---+-----------------------------------------------+|XIAOMEI|21|2022-02-1000:09:55||XIAOMING|22|2022-02-1000:10:07||XIAOXUE|23|2022-02-1000:10:15|+--------------------+---+-----------------------------------------------+6.4.2

SQL語法風(fēng)格從RDD轉(zhuǎn)換得到DataFrame提綱12使用編程方式定義RDD模式利用反射機(jī)制推斷RDD模式6.5.1利用反射機(jī)制推斷RDD模式“/usr/local/spark/examples/src/main/resources/”目錄下Michael,29Andy,30Justin,19people.txt6.5.1利用反射機(jī)制推斷RDD模式Michael,29Andy,30Justin,19DataFrame內(nèi)存加載生成people.txt6.5.1利用反射機(jī)制推斷RDD模式RDD利用反射機(jī)制推斷RDD的模式6.5.1利用反射機(jī)制推斷RDD模式

首先需定義一個(gè)caseclass，只有caseclass才能被Spark隱式地轉(zhuǎn)換為DataFramescala>importorg.apache.spark.sql.catalyst.encoders.ExpressionEncoderimportorg.apache.spark.sql.catalyst.encoders.ExpressionEncoderscala>importorg.apache.spark.sql.Encoderimportorg.apache.spark.sql.Encoderscala>importspark.implicits._//導(dǎo)入包，支持把一個(gè)RDD隱式轉(zhuǎn)換為一個(gè)DataFrameimportspark.implicits._6.5.1利用反射機(jī)制推斷RDD模式scala>caseclassPerson(name:String,age:Long)//定義一個(gè)caseclassdefinedclassPersonscala>valpeopleDF=spark.sparkContext.|textFile("file:///usr/local/spark/examples/src/main/resources/people.txt").|map(_.split(",")).|map(attributes=>Person(attributes(0),attributes(1).trim.toInt)).toDF()peopleDF:org.apache.spark.sql.DataFrame=[name:string,age:bigint]

首先需定義一個(gè)caseclass，只有caseclass才能被Spark隱式地轉(zhuǎn)換為DataFrame6.5.1利用反射機(jī)制推斷RDD模式scala>peopleDF.createOrReplaceTempView("people")//必須注冊(cè)為臨時(shí)表才能供下面的查詢使用scala>valpersonsRDD=spark.sql("selectname,agefrompeoplewhereage>20")//最終生成一個(gè)DataFrame，下面是系統(tǒng)執(zhí)行返回的信息personsRDD:org.apache.spark.sql.DataFrame=[name:string,age:bigint]scala>personsRDD.map(t=>"Name:"+t(0)+","+"Age:"+t(1)).show()//DataFrame中的每個(gè)元素都是一行記錄，包含name和age兩個(gè)字段，分別用t(0)和t(1)來獲取值//下面是系統(tǒng)執(zhí)行返回的信息+------------------+|value|+------------------+|Name:Michael,Age:29||Name:Andy,Age:30|+------------------+6.5.2使用編程方式定義RDD模式采用編程方式定義RDD模式?6.5.2使用編程方式定義RDD模式提前定義caseclassname字段age字段6.5.2使用編程方式定義RDD模式構(gòu)建關(guān)系表事先不能夠知道字段后面通過動(dòng)態(tài)的方式得到信息6.5.2使用編程方式定義RDD模式采用編程方式定義RDD模式第二種方式6.5.2使用編程方式定義RDD模式磁盤people.txtMichael,29Andy,30Justin,19加載進(jìn)來DataFrame進(jìn)行SQL查詢6.5.2使用編程方式定義RDD模式6.5.2使用編程方式定義RDD模式模式信息也叫表頭什么類型？什么類型？是否為空？6.5.2使用編程方式定義RDD模式什么類型？是否為空？模式信息幾個(gè)字段？字段類型？是否為空？6.5.2使用編程方式定義RDD模式三條記錄從文本加載進(jìn)來6.5.2使用編程方式定義RDD模式6.5.2使用編程方式定義RDD模式scala>importorg.apache.spark.sql.types._importorg.apache.spark.sql.types._scala>importorg.apache.spark.sql.Rowimportorg.apache.spark.sql.Row//生成字段scala>valfields=Array(StructField("name",StringType,true),StructField("age",IntegerType,true))fields:Array[org.apache.spark.sql.types.StructField]=Array(StructField(name,StringType,true),StructField(age,IntegerType,true))scala>valschema=StructType(fields)schema:org.apache.spark.sql.types.StructType=StructType(StructField(name,StringType,true),StructField(age,IntegerType,true))//從上面信息可以看出，schema描述了模式信息，模式中包含name和age兩個(gè)字段//shcema就是“表頭”6.5.2使用編程方式定義RDD模式//下面加載文件生成RDDscala>valpeopleRDD=spark.sparkContext.|textFile("file:///usr/local/spark/examples/src/main/resources/people.txt")peopleRDD:org.apache.spark.rdd.RDD[String]=file:///usr/local/spark/examples/src/main/resources/people.txtMapPartitionsRDD[1]attextFileat<console>:26//對(duì)peopleRDD這個(gè)RDD中的每一行元素都進(jìn)行解析scala>valrowRDD=peopleRDD.map(_.split(",")).|map(attributes=>Row(attributes(0),attributes(1).trim.toInt))rowRDD:org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]=MapPartitionsRDD[3]atmapat<console>:29//上面得到的rowRDD就是“表中的記錄”6.5.2使用編程方式定義RDD模式//下面把“表頭”和“表中的記錄”拼裝起來scala>valpeopleDF=spark.createDataFrame(rowRDD,schema)peopleDF:org.apache.spark.sql.DataFrame=[name:string,age:int]//必須注冊(cè)為臨時(shí)表才能供下面查詢使用scala>peopleDF.createOrReplaceTempView("people")scala>valresults=spark.sql("SELECTname,ageFROMpeople")results:org.apache.spark.sql.DataFrame=[name:string,age:int]

scala>results.|map(attributes=>"name:"+attributes(0)+","+"age:"+attributes(1)).|show()+--------------------+|value|+--------------------+|name:Michael,age:29||name:Andy,age:30||name:Justin,age:19|+--------------------+使用SparkSQL讀寫數(shù)據(jù)庫(kù)6.6使用SparkSQL讀寫數(shù)據(jù)庫(kù)SparkSQL

準(zhǔn)備工作

01

編寫?yīng)毩?yīng)用

程序訪問MySQL04讀取MySQL數(shù)據(jù)庫(kù)中的數(shù)據(jù)02

向MySQL數(shù)據(jù)庫(kù)寫入數(shù)據(jù)036.6.1準(zhǔn)備工作請(qǐng)參考廈門大學(xué)數(shù)據(jù)庫(kù)實(shí)驗(yàn)室博客教程《Ubuntu安裝MySQL》教程地址：

/blog/install-mysql/，在Linux系統(tǒng)中安裝好MySQL數(shù)據(jù)庫(kù)6.6.1準(zhǔn)備工作

輸入下面SQL語句完成數(shù)據(jù)庫(kù)和表的創(chuàng)建

$servicemysqlstart$mysql-uroot-p#屏幕會(huì)提示你輸入密碼

在Linux中啟動(dòng)MySQL數(shù)據(jù)庫(kù)

mysql>createdatabasespark;mysql>usespark;mysql>createtablestudent(idint(4),namechar(20),genderchar(4),ageint(4));mysql>insertintostudentvalues(1,'Xueqian','F',23);mysql>insertintostudentvalues(2,'Weiliang','M',24);mysql>select*fromstudent;6.6.1準(zhǔn)備工作010203下載驅(qū)動(dòng)程序下載MySQL的JDBC驅(qū)動(dòng)程序，比如mysql-connector-java-5.1.40.tar.gz拷貝到安裝目錄把該驅(qū)動(dòng)程序拷貝到spark

的安裝目錄下”/usr/local/spark/jars”啟動(dòng)spark-shell啟動(dòng)spark-shell，啟動(dòng)SparkShell時(shí)，必須指定mysql連接驅(qū)動(dòng)jar包6.6.1準(zhǔn)備工作

準(zhǔn)備工作：下載驅(qū)動(dòng)程序，拷貝到安裝目錄，啟動(dòng)spark-shell

$cd/usr/local/spark$./bin/spark-shell\--jars/usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar\--driver-class-path/usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar6.6.1準(zhǔn)備工作scala>valjdbcDF=spark.read.format("jdbc").|option("url","jdbc:mysql://localhost:3306/spark").|option("driver","com.mysql.jdbc.Driver").|option("dbtable","student").|option("user","root").|option("password","123456").|load()scala>jdbcDF.show()+---+--------+------+---+|id|name|gender|age|+---+--------+------+---+|1|Xueqian|F|23||2|Weiliang|M|24|+---+--------+------+---+

執(zhí)行以下命令連接數(shù)據(jù)庫(kù)，讀取數(shù)據(jù)，并顯示6.6.3

向MySQL數(shù)據(jù)庫(kù)寫入數(shù)據(jù)

在MySQL數(shù)據(jù)庫(kù)中創(chuàng)建了一個(gè)名稱為spark的數(shù)據(jù)庫(kù)，查看一下數(shù)據(jù)庫(kù)內(nèi)容importjava.util.Propertiesimportorg.apache.spark.sql.types._importorg.apache.spark.sql.Row

//下面我們?cè)O(shè)置兩條數(shù)據(jù)表示兩個(gè)學(xué)生信息valstudentRDD=spark.sparkContext.parallelize(Array("3RongchengM26","4GuanhuaM27")).map(_.split(""))

//下面要設(shè)置模式信息valschema=StructType(List(StructField("id",IntegerType,true),StructField("name",StringType,true),StructField("gender",StringType,true),StructField("age",IntegerType,true)))

在spark-shell中編寫程序，往spark.student表中插入兩條記錄6.6.3

向MySQL數(shù)據(jù)庫(kù)寫入數(shù)據(jù)6.6.3

向MySQL數(shù)據(jù)庫(kù)寫入數(shù)據(jù)//下面創(chuàng)建Row對(duì)象，每個(gè)Row對(duì)象都是rowRDD中的一行valrowRDD=studentRDD.map(p=>Row(p(0).toInt,p(1).trim,p(2).trim,p(3).toInt))

//建立起Row對(duì)象和模式之間的對(duì)應(yīng)關(guān)系，也就是把數(shù)據(jù)和模式對(duì)應(yīng)起來valstudentDF=spark.createDataFrame(rowRDD,schema)

//下面創(chuàng)建一個(gè)prop變量用來保存JDBC連接參數(shù)valprop=newProperties()prop.put("user","root")//表示用戶名是rootprop.put("password","hadoop")//表示密碼是hadoopprop.put("driver","com.mysql.jdbc.Driver")//表示驅(qū)動(dòng)程序是com.mysql.jdbc.Driver

//下面就可以連接數(shù)據(jù)庫(kù)，采用append模式，表示追加記錄到數(shù)據(jù)庫(kù)spark的student表中studentDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/spark","spark.student",prop)mysql>select*fromstudent;+------+-----------+--------+------+|id|name|gender|age|+------+-----------+--------+------+|1|Xueqian|F|23||2|Weiliang|M|24||3|Rongcheng|M|26||4|Guanhua|M|27|+------+-----------+--------+------+4rowsinset(0.00sec)

查看MySQL數(shù)據(jù)庫(kù)中的spark.student表發(fā)生了什么變化6.6.3

向MySQL數(shù)據(jù)庫(kù)寫入數(shù)據(jù)6.6.4編寫?yīng)毩?yīng)用程序訪問MySQL1.

讀取MySQLimportorg.apache.log4j.{Level,Logger}importorg.apache.spark.sql.SparkSessionobjectSparkReadMySQL{defmain(args:Array[String]):Unit={Logger.getLogger("org").setLevel(Level.ERROR)valspark=SparkSession.builder().appName("SparkReadMySQL").getOrCreate()valdf=spark.read.format("jdbc").option("url","jdbc:mysql://localhost:3306/spark").option("driver","com.mysql.jdbc.Driver").option("dbtable","student").option("user","root").option("password","123456").load()df.show()spark.stop()}}

$/usr/local/spark/bin/spark-submit\>--jars\>/usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar\>--class"SparkReadMySQL"\>/home/hadoop/sparkapp/target/scala-2.12/simple-project_2.12-1.0.jar

對(duì)代碼進(jìn)行編譯打包，然后執(zhí)行如下命令運(yùn)行程序6.6.4編寫?yīng)毩?yīng)用程序訪問MySQL6.6.4編寫?yīng)毩?yīng)用程序訪問MySQL2.

寫入MySQLimportjava.util.Propertiesimportorg.apache.spark.sql.types._importorg.apache.spark.sql.Rowimportorg.apache.log4j.{Level,Logger}importorg.apache.spark.sql.SparkSessionobjectSparkWriteMySQL{defmain(args:Array[String]):Unit={Logger.getLogger("org").setLevel(Level.ERROR)valspark=SparkSession.builder().appName("SparkWriteMySQL").getOrCreate()//下面我們?cè)O(shè)置兩條數(shù)據(jù)表示兩個(gè)學(xué)生信息valstudentRDD=spark.sparkContext.parallelize(Array("3RongchengM26","4GuanhuaM27")).map(_.split(""))6.6.4編寫?yīng)毩?yīng)用程序訪問MySQL

//下面要設(shè)置模式信息valschema=StructType(List(StructField("id",IntegerType,true),StructField("name",StringType,true),StructField("gender",StringType,true),StructField("age",IntegerType,true)))

//下面創(chuàng)建Row對(duì)象，每個(gè)Row對(duì)象都是rowRDD中的一行valrowRDD=studentRDD.map(p=>Row(p(0).toInt,p(1).trim,p(2).trim,p(3).toInt))

//建立起Row對(duì)象和模式之間的對(duì)應(yīng)關(guān)系，也就是把數(shù)據(jù)和模式對(duì)應(yīng)起來valstudentDF=spark.createDataFrame(rowRDD,schema)

//下面創(chuàng)建一個(gè)prop變量用來保存JDBC連接參數(shù)valprop=newProperties()prop.put("user","root")//表示用戶名是rootprop.put("password","123456")//表示密碼是123456prop.put("driver","com.mysql.jdbc.Driver")//表示驅(qū)動(dòng)程序是com.mysql.jdbc.Driver

//下面就可以連接數(shù)據(jù)庫(kù)，采用append模式，表示追加記錄到數(shù)據(jù)庫(kù)spark的student表中studentDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/spark","spark.student",prop)spark.stop()}}$/usr/local/spark/bin/spark-submit\>--jars\>/usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar\>--class"SparkWriteMySQL"\>/home/hadoop/sparkapp/target/scala-2.12/simple-project_2.12-1.0.jar

對(duì)代碼進(jìn)行編譯打包，然后執(zhí)行如下命令運(yùn)行程序6.6.4編寫?yīng)毩?yīng)用程序訪問MySQLDataSet6.7DataSetDataFrame、DataSet和RDD的區(qū)別創(chuàng)建DataSetRDD、DataFrame和DataSet之間的相互轉(zhuǎn)換詞頻統(tǒng)計(jì)實(shí)例目錄CONTENT6.7.1DataFrame、DataSet和RDD的區(qū)別RDDDataSetDataFrame6.7.1DataFrame、DataSet和RDD的區(qū)別

圖

RDD中的數(shù)據(jù)保存方式

圖

DataFrame中的數(shù)據(jù)保存方式6.7.1DataFrame、DataSet和RDD的區(qū)別

圖

DataSet中的數(shù)據(jù)保存方式之一

圖

DataSet中的數(shù)據(jù)保存方式之二6.7.1DataFrame、DataSet和RDD的區(qū)別RDDDataFrameDataSet不可變性是是是分區(qū)是是是模式?jīng)]有有有查詢優(yōu)化器沒有有有API級(jí)別低高高是否類型安全是否是何時(shí)檢測(cè)語法錯(cuò)誤編譯時(shí)編譯時(shí)編譯時(shí)何時(shí)檢測(cè)分析錯(cuò)誤編譯時(shí)運(yùn)行時(shí)編譯時(shí)6.7.1DataFrame、DataSet和RDD的區(qū)別圖SparkSQL中的查詢優(yōu)化6.7.1DataFrame、DataSet和RDD的區(qū)別（1）沒有針對(duì)特殊場(chǎng)景進(jìn)行優(yōu)化，比如對(duì)于結(jié)構(gòu)化數(shù)據(jù)處理相對(duì)于SQL來比顯得非常麻煩（2）默認(rèn)采用的是Java序列化方式，序列化結(jié)果比較大，而且數(shù)據(jù)存儲(chǔ)在Java堆內(nèi)存中，導(dǎo)致垃圾回收比較頻繁（1）相比于傳統(tǒng)的MapReduce框架，Spark在RDD中內(nèi)置了很多函數(shù)操作，如map、filter、sort等,方便處理結(jié)構(gòu)化或非結(jié)構(gòu)化數(shù)據(jù)（2）面向?qū)ο缶幊?，直接存?chǔ)Java對(duì)象，類型轉(zhuǎn)化比較安全RDD缺點(diǎn)RDD優(yōu)點(diǎn)6.7.1DataFrame、DataSet和RDD的區(qū)別（1）結(jié)構(gòu)化數(shù)據(jù)處理非常方便，支持Avro、CSV、Elasticsearch、Cassandra等類型數(shù)據(jù)，也支持Hive、MySQL等傳統(tǒng)數(shù)據(jù)表；（2）可以進(jìn)行有針對(duì)性的優(yōu)化，比如采用Kryo序列化，由于Spark中已經(jīng)保存了數(shù)據(jù)結(jié)構(gòu)元信息，因此，序列化時(shí)就不需要帶上元信息，這就大大減少了序列化開銷，而且數(shù)據(jù)保存在堆外內(nèi)存中，減少了垃圾回收次數(shù)，所以運(yùn)行更快DataFrame優(yōu)點(diǎn)（1）不支持編譯時(shí)類型安全，運(yùn)行時(shí)才能確定是否有問題；（2）對(duì)于對(duì)象支持不友好，RDD內(nèi)部數(shù)據(jù)直接以Java對(duì)象存儲(chǔ)，而DataFrame內(nèi)存存儲(chǔ)的是Row對(duì)象，而不是自定義對(duì)象DataFrame缺點(diǎn)6.7.1DataFrame、DataSet和RDD的區(qū)別04030102和DataFrame一樣DataSet整合了RDD和DataFrame的優(yōu)點(diǎn)和RDD一樣Dataset支持結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)DataSet支持結(jié)構(gòu)化數(shù)據(jù)的SQL查詢采用堆外內(nèi)存存儲(chǔ)垃圾回收比較高效DataSet支持自定義對(duì)象存儲(chǔ)6.7.1DataFrame、DataSet和RDD的區(qū)別01則使用DataFrame或DataSet如需豐富的語義、高層次的抽象和特定API02或lambda函數(shù)，則用DataFrame或DataSet如果處理要求涉及到filter、map查詢04則使用DataFrame或DataSet如果想統(tǒng)一和簡(jiǎn)化Spark的API03需要類型化的JVM對(duì)象，并希望利用Tungsten編碼進(jìn)行高效的序列化和反序列化，則使用DataSet如需在編譯時(shí)獲得更高的類型安全性05則使用DataFrame如果與R語言或Python語言結(jié)合使用06盡量使用RDD如果需要更多的控制功能6.7.1DataFrame、DataSet和RDD的區(qū)別基于RDD的結(jié)構(gòu)化數(shù)據(jù)抽象提供的DataFrame

和

DataSetSQL6.7.2

創(chuàng)建DataSetscala>valds1=spark.createDataset(1to5)ds1:org.apache.spark.sql.Dataset[Int]=[value:int]scala>ds1.show()+--------+|value|+--------+|1||2||3||4||5|+--------+

1.

使用createDataset方法創(chuàng)建6.7.2

創(chuàng)建DataSetscala>valds2=spark.createDataset(sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt"))ds2:org.apache.spark.sql.Dataset[String]=[value:string]scala>ds2.show()+---------------+|value|+---------------+|Michael,29||Andy,30||Justin,19|+---------------+

1.

使用createDataset方法創(chuàng)建6.7.2

創(chuàng)建DataSetscala>caseclassPerson(name:String,age:Int)definedclassPersonscala>valdata=List(Person("ZhangSan",23),Person("LiSi",35))data:List[Person]=List(Person(ZhangSan,23),Person(LiSi,35))scala>valds3=data.toDSds3:org.apache.spark.sql.Dataset[Person]=[name:string,age:int]scala>ds3.show()+--------------+-----+|name|age|+--------------+-----+|ZhangSan|23||LiSi|35|+--------------+----+

2.

通過toDS方法生成DataSet6.7.2

創(chuàng)建DataSetscala>caseclassPerson(name:String,age:Long)definedclassPersonscala>val