《Spark編程基礎(chǔ)Python版》PPT課件

1、Spark編程基礎(chǔ)(Python版)廈門大學(xué)計算機(jī)科學(xué)系 2020版第第5章章 Spark SQL（PPT版本號：版本號：2020年年4月版）月版） Spark編程基礎(chǔ)（編程基礎(chǔ)（Python版）版）教材官網(wǎng)： http:/ Spark SQL簡介簡介5.2 DataFrame概述概述5.3 DataFrame的創(chuàng)建的創(chuàng)建5.4 DataFrame的保存的保存5.5 DataFrame的常用操作的常用操作5.6 從從RDD轉(zhuǎn)換得到轉(zhuǎn)換得到DataFrame5.7 使用使用Spark SQL讀寫數(shù)據(jù)庫讀寫數(shù)據(jù)庫百度搜索廈門大學(xué)數(shù)據(jù)庫實驗室網(wǎng)站訪問平臺Spark編程基礎(chǔ)(Python版)5.1 S

2、park SQL簡介5.1.1 從Shark說起5.1.2 Spark SQL設(shè)計5.1.3 為什么推出Spark SQLSpark編程基礎(chǔ)(Python版)5.1.1 從Shark說起 Hive中SQL查詢的MapReduce作業(yè)轉(zhuǎn)化過程Hive: SQL-on-HadoopSpark編程基礎(chǔ)(Python版)5.1.1 從Shark說起Shark即Hive on Spark，為了實現(xiàn)與Hive兼容，Shark在HiveQL方面重用了Hive中HiveQL的解析、邏輯執(zhí)行計劃翻譯、執(zhí)行計劃優(yōu)化等邏輯，可以近似認(rèn)為僅將物理執(zhí)行計劃從MapReduce作業(yè)替換成了Spark作業(yè)，通過Hive的H

3、iveQL解析，把HiveQL翻譯成Spark上的RDD操作Spark編程基礎(chǔ)(Python版)5.1.1 從Shark說起Shark的出現(xiàn)，使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高Spark編程基礎(chǔ)(Python版)5.1.1 從Shark說起Shark的設(shè)計導(dǎo)致了兩個問題：一是執(zhí)行計劃優(yōu)化完全依賴于Hive，不方便添加新的優(yōu)化策略二是因為Spark是線程級并行，而MapReduce是進(jìn)程級并行，因此，Spark在兼容Hive的實現(xiàn)上存在線程安全問題，導(dǎo)致Shark不得不使用另外一套獨立維護(hù)的打了補(bǔ)丁的Hive源碼分支Spark編程基礎(chǔ)(Python版)5.1.

4、1 從Shark說起2014年6月1日Shark項目和Spark SQL項目的主持人Reynold Xin宣布：停止對Shark的開發(fā)，團(tuán)隊將所有資源放在Spark SQL項目上，至此，Shark的發(fā)展畫上了句號，但也因此發(fā)展出兩個分支：Spark SQL和Hive on SparkSpark SQL作為Spark生態(tài)的一員繼續(xù)發(fā)展，而不再受限于Hive，只是兼容HiveHive on Spark是一個Hive的發(fā)展計劃，該計劃將Spark作為Hive的底層引擎之一，也就是說，Hive將不再受限于一個引擎，可以采用Map-Reduce、Tez、Spark等引擎Spark編程基礎(chǔ)(Python版

5、)5.1.2 Spark SQL設(shè)計Spark SQL在Hive兼容層面僅依賴HiveQL解析、Hive元數(shù)據(jù)，也就是說，從HQL被解析成抽象語法樹（AST）起，就全部由Spark SQL接管了。Spark SQL執(zhí)行計劃生成和優(yōu)化都由Catalyst（函數(shù)式關(guān)系查詢優(yōu)化框架）負(fù)責(zé)圖 Spark SQL架構(gòu)Spark編程基礎(chǔ)(Python版)5.1.2 Spark SQL設(shè)計圖 Spark SQL支持的數(shù)據(jù)格式和編程語言Spark SQL增加了DataFrame（即帶有Schema信息的RDD），使用戶可以在Spark SQL中執(zhí)行SQL語句，數(shù)據(jù)既可以來自RDD，也可以是Hive、HDFS、

6、Cassandra等外部數(shù)據(jù)源，還可以是JSON格式的數(shù)據(jù)Spark SQL目前支持Scala、Java、Python三種語言，支持SQL-92規(guī)范Spark編程基礎(chǔ)(Python版)5.1.3 為什么推出Spark SQLSpark編程基礎(chǔ)(Python版)5.1.3 為什么推出Spark SQLSpark編程基礎(chǔ)(Python版)5.1.3 為什么推出Spark SQLSpark SQL填補(bǔ)了這個鴻溝：首先，可以提供DataFrame API，可以對內(nèi)部和外部各種數(shù)據(jù)源執(zhí)行各種關(guān)系型操作其次，可以支持大數(shù)據(jù)中的大量數(shù)據(jù)源和數(shù)據(jù)分析算法Spark SQL可以融合：傳統(tǒng)關(guān)系數(shù)據(jù)庫的結(jié)構(gòu)化數(shù)據(jù)管

7、理能力和機(jī)器學(xué)習(xí)算法的數(shù)據(jù)處理能力關(guān)系數(shù)據(jù)庫已經(jīng)很流行關(guān)系數(shù)據(jù)庫在大數(shù)據(jù)時代已經(jīng)不能滿足要求首先，用戶需要從不同數(shù)據(jù)源執(zhí)行各種操作，包括結(jié)構(gòu)化、半結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)其次，用戶需要執(zhí)行高級分析，比如機(jī)器學(xué)習(xí)和圖像處理在實際大數(shù)據(jù)應(yīng)用中，經(jīng)常需要融合關(guān)系查詢和復(fù)雜分析算法（比如機(jī)器學(xué)習(xí)或圖像處理），但是，缺少這樣的系統(tǒng)Spark編程基礎(chǔ)(Python版)5.2 DataFrame概述DataFrame的推出，讓Spark具備了處理大規(guī)模結(jié)構(gòu)化數(shù)據(jù)的能力，不僅比原有的RDD轉(zhuǎn)化方式更加簡單易用，而且獲得了更高的計算性能Spark能夠輕松實現(xiàn)從MySQL到DataFrame的轉(zhuǎn)化，并且支持SQL查詢

8、圖 DataFrame與RDD的區(qū)別RDD是分布式的 Java對象的集合，但是，對象內(nèi)部結(jié)構(gòu)對于RDD而言卻是不可知的DataFrame是一種以RDD為基礎(chǔ)的分布式數(shù)據(jù)集，提供了詳細(xì)的結(jié)構(gòu)信息Spark編程基礎(chǔ)(Python版)5.3 DataFrame的創(chuàng)建從Spark2.0以上版本開始，Spark使用全新的SparkSession接口替代Spark1.6中的SQLContext及HiveContext接口來實現(xiàn)其對數(shù)據(jù)加載、轉(zhuǎn)換、處理等功能。SparkSession實現(xiàn)了SQLContext及HiveContext所有功能SparkSession支持從不同的數(shù)據(jù)源加載數(shù)據(jù)，并把數(shù)據(jù)轉(zhuǎn)換成D

9、ataFrame，并且支持把DataFrame轉(zhuǎn)換成SQLContext自身中的表，然后使用SQL語句來操作數(shù)據(jù)。SparkSession亦提供了HiveQL以及其他依賴于Hive的功能的支持可以通過如下語句創(chuàng)建一個SparkSession對象：from pyspark import SparkContext,SparkConffrom pyspark.sql import SparkSessionspark = SparkSession.builder.config(conf = SparkConf().getOrCreate()實際上，在啟動進(jìn)入pyspark以后，pyspark就默認(rèn)提供

10、了一個SparkContext對象（名稱為sc）和一個SparkSession對象（名稱為spark）Spark編程基礎(chǔ)(Python版)5.3 DataFrame的創(chuàng)建在創(chuàng)建DataFrame時，可以使用spark.read操作，從不同類型的文件中加載數(shù)據(jù)創(chuàng)建DataFrame，例如：spark.read.text(people.txt)：讀取文本文件people.txt創(chuàng)建DataFramespark.read.json(people.json)：讀取people.json文件創(chuàng)建DataFrame；在讀取本地文件或HDFS文件時，要注意給出正確的文件路徑spark.read.parque

11、t(“people.parquet”)：讀取people.parquet文件創(chuàng)建DataFrameSpark編程基礎(chǔ)(Python版)5.3 DataFrame的創(chuàng)建或者也可以使用如下格式的語句：spark.read.format(text).load(people.txt)：讀取文本文件people.json創(chuàng)建DataFrame；spark.read.format(json).load(people.json)：讀取JSON文件people.json創(chuàng)建DataFrame；spark.read.format(parquet).load(people.parquet)：讀取Parquet文件

12、people.parquet創(chuàng)建DataFrame。Spark編程基礎(chǔ)(Python版)5.3 DataFrame的創(chuàng)建在“/usr/local/spark/examples/src/main/resources/”這個目錄下，這個目錄下有兩個樣例數(shù)據(jù)people.json和people.txt。people.json文件的內(nèi)容如下：name:Michaelname:Andy, age:30name:Justin, age:19people.txt文件的內(nèi)容如下：Michael, 29Andy, 30Justin, 19一個實例一個實例Spark編程基礎(chǔ)(Python版)5.3 DataFra

13、me的創(chuàng)建 df = spark.read.json(file:/usr/local/spark/examples/src/main/resources/people.json) df.show()+-+-+| age| name|+-+-+|null|Michael| 30| Andy| 19| Justin|+-+-+Spark編程基礎(chǔ)(Python版)5.4 DataFrame的保存可以使用spark.write操作，把一個DataFrame保存成不同格式的文件，例如，把一個名稱為df的DataFrame保存到不同格式文件中，方法如下：df.write.text(people.txt)d

14、f.write.json(people.json“)df.write.parquet(people.parquet“)或者也可以使用如下格式的語句：df.write.format(text).save(people.txt)df.write.format(json).save(people.json)df.write.format (parquet).save(people.parquet)Spark編程基礎(chǔ)(Python版)5.4 DataFrame的保存下面從示例文件people.json中創(chuàng)建一個DataFrame，名稱為peopleDF，把peopleDF保存到另外一個JSON文件中，

15、然后，再從peopleDF中選取一個列（即name列），把該列數(shù)據(jù)保存到一個文本文件中 peopleDF = spark.read.format(json). load(file:/usr/local/spark/examples/src/main/resources/people.json) peopleDF.select(name, age).write.format(json). save(file:/usr/local/spark/mycode/sparksql/newpeople.json) peopleDF.select(name).write.format(text). save

16、(file:/usr/local/spark/mycode/sparksql/newpeople.txt)會新生成一個名稱為newpeople.json的目錄（不是文件）和一個名稱為newpeople.txt的目錄（不是文件）part-00000-3db90180-ec7c-4291-ad05-df8e45c77f4d.json_SUCCESSSpark編程基礎(chǔ)(Python版)5.5 DataFrame的常用操作可以執(zhí)行一些常用的DataFrame操作printSchema()select() df=spark.read.json(“people.json”)Spark編程基礎(chǔ)(Python

17、版)5.5 DataFrame的常用操作filter()groupBy()Spark編程基礎(chǔ)(Python版)5.5 DataFrame的常用操作sort()Spark編程基礎(chǔ)(Python版)5.6 從RDD轉(zhuǎn)換得到DataFrame5.6.1 利用反射機(jī)制推斷RDD模式5.6.2 使用編程方式定義RDD模式Spark編程基礎(chǔ)(Python版)5.6.1 利用反射機(jī)制推斷RDD模式在“/usr/local/spark/examples/src/main/resources/”目錄下，有個Spark安裝時自帶的樣例數(shù)據(jù)people.txt，其內(nèi)容如下：Michael, 29Andy, 30Ju

18、stin, 19現(xiàn)在要把people.txt加載到內(nèi)存中生成一個DataFrame，并查詢其中的數(shù)據(jù)Spark編程基礎(chǔ)(Python版)5.6.1 利用反射機(jī)制推斷RDD模式 from pyspark.sql import Row people = spark.sparkContext. textFile(file:/usr/local/spark/examples/src/main/resources/people.txt). map(lambda line: line.split(,). map(lambda p: Row(name=p0, age=int(p1) schemaPeople

19、 = spark.createDataFrame(people)#必須注冊為臨時表才能供下面的查詢使用 schemaPeople.createOrReplaceTempView(people) personsDF = spark.sql(select name,age from people where age 20)#DataFrame中的每個元素都是一行記錄，包含name和age兩個字段，分別用和p.age來獲取值 personsRDD=personsDF.rdd.map(lambda p:Name: ++ ,+Age: +str(p.age) personsRDD

20、.foreach(print)Name: Michael,Age: 29Name: Andy,Age: 30Spark編程基礎(chǔ)(Python版)5.6.1 利用反射機(jī)制推斷RDD模式Spark編程基礎(chǔ)(Python版)5.6.2 使用編程方式定義RDD模式當(dāng)無法提前獲知數(shù)據(jù)結(jié)構(gòu)時，就需要采用編程方式定義RDD模式。比如，現(xiàn)在需要通過編程方式把people.txt加載進(jìn)來生成DataFrame，并完成SQL查詢。圖 通過編程方式定義RDD模式的實現(xiàn)過程Spark編程基礎(chǔ)(Python版)5.6.2 使用編程方式定義RDD模式 from pyspark.sql.types import * fro

21、m pyspark.sql import Row#下面生成“表頭” schemaString = name age fields = StructField(field_name, StringType(), True) for field_name in schemaString.split( ) schema = StructType(fields)#下面生成“表中的記錄” lines = spark.sparkContext. textFile(file:/usr/local/spark/examples/src/main/resources/people.txt) parts = li

22、nes.map(lambda x: x.split(,) people = parts.map(lambda p: Row(p0, p1.strip()#下面把“表頭”和“表中的記錄”拼裝在一起 schemaPeople = spark.createDataFrame(people, schema)剩余代碼見下一頁Spark編程基礎(chǔ)(Python版)5.6.2 使用編程方式定義RDD模式Spark編程基礎(chǔ)(Python版)5.6.2 使用編程方式定義RDD模式#注冊一個臨時表供下面查詢使用 schemaPeople.createOrReplaceTempView(people) results

23、 = spark.sql(SELECT name,age FROM people) results.show()+-+-+| name|age|+-+-+|Michael| 29| Andy| 30| Justin| 19|+-+-+Spark編程基礎(chǔ)(Python版)5.7 使用Spark SQL讀寫數(shù)據(jù)庫Spark SQL可以支持Parquet、JSON、Hive等數(shù)據(jù)源，并且可以通過JDBC連接外部數(shù)據(jù)源5.7.1 準(zhǔn)備工作5.7.2 讀取MySQL數(shù)據(jù)庫中的數(shù)據(jù)5.7.3 向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)Spark編程基礎(chǔ)(Python版)5.7.1 準(zhǔn)備工作請參考廈門大學(xué)數(shù)據(jù)庫實驗室博客教

24、程 Ubuntu安裝MySQL ，在Linux系統(tǒng)中安裝好MySQL數(shù)據(jù)庫教程地址： http:/ 準(zhǔn)備工作在Linux中啟動MySQL數(shù)據(jù)庫$ service mysql start$ mysql -u root -p#屏幕會提示你輸入密碼mysql create database spark;mysql use spark;mysql create table student (id int(4), name char(20), gender char(4), age int(4);mysql insert into student values(1,Xueqian,F,23);mysql

25、 insert into student values(2,Weiliang,M,24);mysql select * from student;輸入下面SQL語句完成數(shù)據(jù)庫和表的創(chuàng)建：Spark編程基礎(chǔ)(Python版)5.7.1 準(zhǔn)備工作下載MySQL的JDBC驅(qū)動程序，比如mysql-connector-java-5.1.40.tar.gz把該驅(qū)動程序拷貝到spark的安裝目錄” /usr/local/spark/jars”下啟動pyspark$ cd /usr/local/spark$ ./bin/pysparkSpark編程基礎(chǔ)(Python版)5.7.2 讀取MySQL數(shù)據(jù)庫中的數(shù)

26、據(jù) jdbcDF = spark.read .format(jdbc) .option(driver,com.mysql.jdbc.Driver) .option(url, jdbc:mysql:/localhost:3306/spark) .option(dbtable, student) .option(user, root) .option(password, 123456) .load() jdbcDF.show()+-+-+-+-+| id| name|gender|age|+-+-+-+-+| 1| Xueqian| F| 23| 2|Weiliang| M| 24|+-+-+-+

27、-+執(zhí)行以下命令連接數(shù)據(jù)庫，讀取數(shù)據(jù)，并顯示：Spark編程基礎(chǔ)(Python版)5.7.3 向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)在MySQL數(shù)據(jù)庫中創(chuàng)建了一個名稱為spark的數(shù)據(jù)庫，并創(chuàng)建了一個名稱為student的表創(chuàng)建后，查看一下數(shù)據(jù)庫內(nèi)容：Spark編程基礎(chǔ)(Python版)5.7.3 向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)現(xiàn)在開始編寫程序，往spark.student表中插入兩條記錄#!/usr/bin/env python3from pyspark.sql import Rowfrom pyspark.sql.types import *from pyspark import SparkContext,SparkConffrom pyspark.sql import SparkSessionspark = SparkSession.builder.config(conf = SparkConf().getOrCreate()#下面設(shè)置模式信息schema = StructType(StructField(id, IntegerType(), True), StructField(name, StringType(), True), StructField(gender, StringType(), True), StructField(age, Int