spark ppt第5章 spark sql

1、Spark SQL匯報(bào)人：鄭宏春日期：2018/09/10目錄1Spark SQL簡介DataFrame與RDD的區(qū)別23DataFrame的創(chuàng)建4從RDD轉(zhuǎn)換得到DataFrame5讀取和保存數(shù)據(jù)從Shark說起輸入Hive: SQL-on-HadoopParserSemantic AnalyzerLogical Plan GeneratorLogical OptimizerPhysical Plan GeneratorPhysical Optimizer輸出Hive中SQL查詢的MapReduce作業(yè)轉(zhuǎn)化過程 選擇最佳的優(yōu)化查詢策略 將邏輯計(jì)劃轉(zhuǎn)成物理計(jì)劃 重寫邏輯查詢計(jì)劃 將查詢塊轉(zhuǎn)換成

2、邏輯查詢計(jì)劃 將抽象語法樹轉(zhuǎn)換成查詢塊 將SQL轉(zhuǎn)換成抽象語法樹從Shark說起 Shark即Hive on Spark，為了實(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操作從Shark說起Shark的出現(xiàn)，使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高從Shark說起Shark的設(shè)計(jì)導(dǎo)致了兩個(gè)問題： 一是執(zhí)行計(jì)劃優(yōu)化完全依賴于Hive，不方便添

3、加新的優(yōu)化策略 二是因?yàn)镾park是線程級并行，而MapReduce是進(jìn)程級并行，因此， Spark在兼容Hive的實(shí)現(xiàn)上存在線程安全問題，導(dǎo)致Shark不得不使用另外一套獨(dú)立維護(hù)的打了補(bǔ)丁的Hive源碼分支從Shark說起2014年6月1日Shark項(xiàng)目和SparkSQL項(xiàng)目的主持人Reynold Xin宣布：停止對Shark的開發(fā)，團(tuán)隊(duì)將所有資源放SparkSQL項(xiàng)目上，至此，Shark的發(fā)展畫上了句話，但也因此發(fā)展出兩個(gè)直線：SparkSQL和Hive on Spark Spark SQL作為Spark生態(tài)的一員繼續(xù)發(fā)展，而不再受限于Hive，只是兼容Hive Hive on Spark

4、是一個(gè)Hive的發(fā)展計(jì)劃，該計(jì)劃將Spark作為Hive的底層引擎之一，也就是說，Hive將不再受限于一個(gè)引擎，可以采用Map-Reduce、Tez、Spark等引擎Spark SQL設(shè)計(jì)Spark SQL在Hive兼容層面僅依賴HiveQL解析、Hive元數(shù)據(jù)，也就是說，從HQL被解析成抽象語法樹（AST）起，就全部由Spark SQL接管了。Spark SQL執(zhí)行計(jì)劃生成和優(yōu)化都由Catalyst（函數(shù)式關(guān)系查詢優(yōu)化框架）負(fù)責(zé)MetastoreSparkHDFS圖 Spark SQL架構(gòu)ClientCLIJDBCDriverCache Mgr.SQLParsercatalystPhysic

5、al PlanExecutionSpark SQL設(shè)計(jì) Spark SQL增加了DataFrame（即帶有Schema信息的RDD），使用戶可以在Spark SQL中執(zhí)行SQL語句，數(shù)據(jù)既可以來自RDD，也可以是Hive、HDFS、Cassandra等外部數(shù)據(jù)源，還可以是JSON格式的數(shù)據(jù) Spark SQL目前支持Scala、Java、Python三種語言，支持SQL-92規(guī)范ScalaPythonJavaHiveQLSQL-92Spark SQL (DataFrame)HDFSHiveCassandraJSONSpark SQL支持的數(shù)據(jù)格式和編程語言圖為什么推出Spark SQL為什么推

6、出Spark SQL為什么推出Spark SQL 關(guān)系數(shù)據(jù)庫已經(jīng)很流行 關(guān)系數(shù)據(jù)庫在大數(shù)據(jù)時(shí)代已經(jīng)不能滿足要求 首先，用戶需要從不同數(shù)據(jù)源執(zhí)行各種操作，包括結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù) 其次，用戶需要執(zhí)行高級分析，比如機(jī)器學(xué)習(xí)和圖像處理 在實(shí)際大數(shù)據(jù)應(yīng)用中，經(jīng)常需要融合關(guān)系查詢和復(fù)雜分析算法（比如機(jī)器學(xué)習(xí)或圖像處理），但是，缺少這樣的系統(tǒng)Spark SQL填補(bǔ)了這個(gè)鴻溝： 首先，可以提供DataFrame API，可以對內(nèi)部和外部各種數(shù)據(jù)源執(zhí)行各種關(guān)系操作 其次，可以支持大量的數(shù)據(jù)源和數(shù)據(jù)分析算法Spark SQL可以融合：傳統(tǒng)關(guān)系數(shù)據(jù)庫的結(jié)構(gòu)化數(shù)據(jù)管理能力和機(jī)器學(xué)習(xí)算法的數(shù)據(jù)處理能力目錄1Spark

7、 SQL簡介DataFrame與RDD的區(qū)別23DataFrame的創(chuàng)建4從RDD轉(zhuǎn)換得到DataFrame5讀取和保存數(shù)據(jù)DataFrame與RDD的區(qū)別 DataFrame的推出，讓Spark具備了處理大規(guī)模結(jié)構(gòu)化數(shù)據(jù)的能力，不僅比原有的RDD轉(zhuǎn)化方式更加簡單易用，而且獲得了更高的計(jì)算性能 Spark能夠輕松實(shí)現(xiàn)從MySQL到DataFrame的轉(zhuǎn)化，并且支持SQL查詢圖 DataFrame與RDD的區(qū)別 RDD是分布式的 Java對象的集合，但是，對象內(nèi)部結(jié)構(gòu)對于RDD而言卻是不可知的 DataFrame是一種以RDD為基礎(chǔ)的分布式數(shù)據(jù)集，提供了詳細(xì)的結(jié)構(gòu)信息DataFrame的創(chuàng)建 從

8、Spark2.0以上版本開始，Spark使用全新的SparkSession接口替代Spark1.6中的SQLContext及HiveContext接口來實(shí)現(xiàn)其對數(shù)據(jù)加載、轉(zhuǎn)換、處理等功能。SparkSession實(shí)現(xiàn)了SQLContext及HiveContext所有功能 SparkSession支持從不同的數(shù)據(jù)源加載數(shù)據(jù)，并把數(shù)據(jù)轉(zhuǎn)換成DataFrame，并且支持把DataFrame轉(zhuǎn)換成SQLContext自身中的表，然后使用SQL語句來操作數(shù)據(jù)。SparkSession亦提供了HiveQL以及其他依賴于Hive的功能的支持目錄1Spark SQL簡介DataFrame與RDD的區(qū)別23Da

9、taFrame的創(chuàng)建4從RDD轉(zhuǎn)換得到DataFrame5讀取和保存數(shù)據(jù)DataFrame的創(chuàng)建 在“/usr/local/spark/examples/src/main/resources/”這個(gè)目錄下，這個(gè)目錄下有兩個(gè)樣例數(shù)據(jù)people.json和people.txt。people.json文件的內(nèi)容如下：people.txt文件的內(nèi)容如下：Michael, 29 Andy, 30 Justin, 19name:Michaelname:Andy, age:30name:Justin, age:19DataFrame的創(chuàng)建scala import org.apache.spark.sql.

10、SparkSession import org.apache.spark.sql.SparkSessionscala val spark=SparkSession.builder().getOrCreate()spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession2bdab835/使支持RDDs轉(zhuǎn)換為DataFrames及后續(xù)sql操作scala import spark.implicits._import spark.implicits._scala val df = spark.read.js

11、on(file:/usr/local/spark/examples/src/main/resources/people.json) df: org.apache.spark.sql.DataFrame = age: bigint, name: stringscala df.show()+ + +| age| name|+ + +|null|Michael| 30| Andy| 19| Justin|+ + +DataFrame的創(chuàng)建可以執(zhí)行一些常用的DataFrame操作DataFrame的創(chuàng)建DataFrame的創(chuàng)建目錄1Spark SQL簡介DataFrame與RDD的區(qū)別23DataFr

12、ame的創(chuàng)建從RDD轉(zhuǎn)換得到DataFrame45讀取和保存數(shù)據(jù)利用反射機(jī)制推斷RDD模式在利用反射機(jī)制推斷RDD模式時(shí)，需要首先定義一個(gè)case class，因?yàn)?，只有case class才能被Spark隱式地轉(zhuǎn)換為DataFrame剩余代碼見下一頁scala case class Person(name: String, age: Long) #定義一個(gè)case class defined class Personscala val peopleDF = spark.sparkContext.textFile(file:/usr/local/spark/examples/src/main/

13、resource s/people.txt).map(_.split(,).map(attributes = Person(attributes(0), attributes(1).trim.toInt).toDF()peopleDF: org.apache.spark.sql.DataFrame = name: string, age: bigintscala import org.apache.spark.sql.catalyst.encoders.ExpressionEncoderimport org.apache.spark.sql.catalyst.encoders.Expressi

14、onEncoderscala import org.apache.spark.sql.Encoder import org.apache.spark.sql.Encoderscala import spark.implicits._ /導(dǎo)入包，支持把一個(gè)RDD隱式轉(zhuǎn)換為一個(gè)DataFrameimport spark.implicits._利用反射機(jī)制推斷RDD模式scala peopleDF.createOrReplaceTempView(“people”) #必須注冊為臨時(shí)表才能供下面的查詢使用scala val personsRDD = spark.sql(select name,age

15、from people where age 20)#最終生成一個(gè)DataFramepersonsRDD: 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)來獲取值+| value|+|Name:Michael,Age:29| Name:Andy,Age:30|+使用編程方式定義RDD模式當(dāng)無法提前定義cas

16、e class時(shí)，就需要采用編程方式定義RDD模式剩余代碼見下一頁scala import org.apache.spark.sql.types._ import org.apache.spark.sql.types._ scala import org.apache.spark.sql.Row import org.apache.spark.sql.Row/生成 RDDscala val peopleRDD = spark.sparkContext.textFile(file:/usr/local/spark/examples/src/main/resour ces/people.txt)p

17、eopleRDD: org.apache.spark.rdd.RDDString = file:/usr/local/spark/examples/src/main/resources/people.txt MapPartitionsRDD1 at textFile at :26使用編程方式定義RDD模式剩余代碼見下一頁scala val schema = StructType(fields)schema: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true), StructFi

18、eld(age,StringType,true)/從上面信息可以看出，schema描述了模式信息，模式中包含name和age兩個(gè)字段/定義一個(gè)模式字符串scala val schemaString = name age schemaString: String = name age/根據(jù)模式字符串生成模式scala val fields = schemaString.split( ).map(fieldName = StructField(fieldName, StringType, nullable = true)fields: Arrayorg.apache.spark.sql.types

19、.StructField = Array(StructField(name,StringType,true), StructField(age,StringType,true)使用編程方式定義RDD模式剩余代碼見下一頁/對peopleRDD 這個(gè)RDD中的每一行元素都進(jìn)行解析scala val rowRDD = peopleRDD.map(_.split(,).map(attributes = Row(attributes(0), attributes(1).trim)rowRDD: org.apache.spark.rdd.RDDorg.apache.spark.sql.Row = MapP

20、artitionsRDD3 at map at :29scala val peopleDF = spark.createDataFrame(rowRDD, schema)peopleDF: org.apache.spark.sql.DataFrame = name: string, age: string使用編程方式定義RDD模式/必須注冊為臨時(shí)表才能供下面查詢使用scala peopleDF.createOrReplaceTempView(people)scala val results = spark.sql(SELECT name,age FROM people) results: or

21、g.apache.spark.sql.DataFrame = name: string, age: stringscala results.map(attributes = name: + attributes(0)+,+age:+attributes(1).show()+| value|+|name: Michael,age:29| name: Andy,age:30| name: Justin,age:19|+把RDD保存成文件第1種保存方法：write.format()支持輸出 json,parquet, jdbc, orc, libsvm, csv, text等格式文件可以看到/usr

22、/local/spark/mycode/這個(gè)目錄下面有個(gè)newpeople.csv文件夾， 包含如下兩個(gè)文件：如果要再次把newpeople.csv中的數(shù)據(jù)加載到RDD中，可以直接使用newpeople.csv目錄名稱，而不需要使用part-r-00000-33184449-cb15- 454c-a30f-9bb43faccac1.csv 文件 scala val textFile = sc.textFile(file:/usr/local/spark/mycode/newpeople.csv)scala val peopleDF = spark.read.format(json).load(

23、file:/usr/local/spark/examples/src/main/resources/people.json) peopleDF: org.apache.spark.sql.DataFrame = age: bigint, name: stringscala peopleDF.select(name, age).write.format(csv).save(file:/usr/local/spark/mycode/newpeople.csv)把RDD保存成文件第2種保存方法：可以看到/usr/local/spark/mycode/這個(gè)目錄下面有個(gè)newpeople.txt文件夾，

24、內(nèi)容如下：如果要再次把newpeople.txt中的數(shù)據(jù)加載到RDD中，可以直接使用newpeople.txt目錄名稱，而不需要使用part-00000文件 scala val textFile = sc.textFile(file:/usr/local/spark/mycode/newpeople.txt)scala val peopleDF = spark.read.format(json).load(file:/usr/local/spark/examples/src/main/resources/people.json) peopleDF: org.apache.spark.sql.D

25、ataFrame = age: bigint, name: stringscala df.rdd.saveAsTextFile(file:/usr/local/spark/mycode/newpeople.txt)目錄1Spark SQL簡介DataFrame與RDD的區(qū)別23DataFrame的創(chuàng)建4從RDD轉(zhuǎn)換得到DataFrame5讀取和保存數(shù)據(jù)讀寫ParquetParquet是一種流行的列式存儲格式，可以高效地存儲具有嵌套字段的記錄。Parquet是語言無關(guān)的，而且不與任何一種數(shù)據(jù)處理框架綁定在一起，適配多種語言和組件，能夠與Parquet配合的組件有： 查詢引擎: Hive, Imp

26、ala, Pig, Presto, Drill, Tajo, HAWQ, IBM Big SQL 計(jì)算框架: MapReduce, Spark, Cascading, Crunch, Scalding, Kite 數(shù)據(jù)模型: Avro, Thrift, Protocol Buffers, POJOs讀寫Parquet從parquet文件中加載數(shù)據(jù)生成DataFrame剩余代碼見下一頁scala import spark.implicits._ import spark.implicits._scala val parquetFileDF = spark.read.parquet(file:/u

27、sr/local/spark/examples/src/main/resources/users.parque t)SLF4J: Failed to load class org.slf4j.impl.StaticLoggerBinder.SLF4J: Defaulting to no-operation (NOP) logger implementationSLF4J: See /codes.html#StaticLoggerBinder for further details. parquetFileDF: org.apache.spark.sql.D

28、ataFrame = name: string, favorite_color: string . 1 more field讀寫Parquetscala parquetFileDF.createOrReplaceTempView(parquetFile)scala val namesDF = spark.sql(SELECT * FROM parquetFile) namesDF: org.apache.spark.sql.DataFrame = name: string, favorite_color: string . 1 more fieldscala namesDF.foreach(a

29、ttributes =println(Name: + attributes(0)+ favorite color:+attributes(1)16/12/02 10:18:49 WARN hadoop.ParquetRecordReader: Can not initialize counter due to context is not a instance of TaskInputOutputContext, but is org.apache.hadoop.mapreduce.task.TaskAttemptContextImplName: Alyssa favorite color:n

30、ull Name: Ben favorite color:red讀寫Parquet將DataFrame保存成parquet文件可以看到”/usr/local/spark/myCode/”這個(gè)目錄下多了一個(gè)newpeople.parquet scala val users = spark.read.parquet(file:/usr/local/spark/myCode/people.parquet)scala import spark.implicits._ import spark.implicits._scala val peopleDF = spark.read.json(file:/u

31、sr/local/spark/examples/src/main/resources/people.json) peopleDF: org.apache.spark.sql.DataFrame = age: bigint, name: stringscala peopleDF.write.parquet(file:/usr/local/spark/mycode/newpeople.parquet)準(zhǔn)備工作在Linux系統(tǒng)中安裝好MySQL數(shù)據(jù)庫 在Linux中啟動MySQL數(shù)據(jù)庫輸入下面SQL語句完成數(shù)據(jù)庫和表的創(chuàng)建：mysql create database spark;mysql use

32、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 insert into student values(2,Weiliang,M,24); mysql select * from student;$ service mysql start$ mysql -u root -p #屏幕會提示你輸入準(zhǔn)備工作 下載MySQL的JDBC驅(qū)動程序，比如mysql-conne

33、ctor-java-5.1.40.tar.gz 把該驅(qū)動程序拷貝到spark的安裝目錄” /usr/local/spark/jars”下 啟動一個(gè)spark-shell，啟動Spark Shell時(shí)，必須指定mysql連接驅(qū)動jar包$ 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-con

34、nector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar讀取MySQL數(shù)據(jù)庫中的數(shù)據(jù)執(zhí)行以下命令連接數(shù)據(jù)庫，讀取數(shù)據(jù)，并顯示：scala val jdbcDF = spark.read.format(jdbc).option(url, jdbc:mysql:/localhost:3306/spark).option(driver,com.mysql.jdbc.Dri ver).option(dbtable, student).option(user,root).option(password, hadoop).load()scala jdbc

35、DF.show()+ + + + +| id| name|gender|age|+ + + + +| 1| Xueqian| F| 23| 2|Weiliang| M| 24|+ + + + +向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)在MySQL數(shù)據(jù)庫中創(chuàng)建了一個(gè)名稱為spark的數(shù)據(jù)庫，并創(chuàng)建了一個(gè)名稱為student的表。創(chuàng)建后，查看一下數(shù)據(jù)庫內(nèi)容：向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)現(xiàn)在開始在spark-shell中編寫程序，往spark.student表中插入兩條記錄import java.util.Propertiesimport org.apache.spark.sql.types._ import or

36、g.apache.spark.sql.Row/下面我們設(shè)置兩條數(shù)據(jù)表示兩個(gè)學(xué)生信息val studentRDD = spark.sparkContext.parallelize(Array(3 Rongcheng M 26,4 Guanhua M 27).map(_.split( )/下面要設(shè)置模式信息val schema = StructType(List(StructField(id, IntegerType, true),StructField(name, StringType, true),StructField(gender, StringType, true),StructFiel

37、d(age, IntegerType, true)/下面創(chuàng)建Row對象，每個(gè)Row對象都是rowRDD中的一行val rowRDD = studentRDD.map(p = Row(p(0).toInt, p(1).trim, p(2).trim, p(3).toInt)/建立起Row對象和模式之間的對應(yīng)關(guān)系，也就是把數(shù)據(jù)和模式對應(yīng)起來val studentDF = spark.createDataFrame(rowRDD, schema)/下面創(chuàng)建一個(gè)prop變量用來保存JDBC連接參數(shù)val prop = new Properties() prop.put(user, root) /表示用

38、戶名是rootprop.put(password, hadoop) /表示是hadoopprop.put(driver,com.mysql.jdbc.Driver) /表示驅(qū)動程序是com.mysql.jdbc.Driver/下面就可以連接數(shù)據(jù)庫，采用append模式，表示追加記錄到數(shù)據(jù)庫spark的student表中studentDF.write.mode(append).jdbc(jdbc:mysql:/localhost:3306/spark, spark.student, prop)向MySQL數(shù)據(jù)庫寫入數(shù)據(jù)可以看一下效果，看看MySQL數(shù)據(jù)庫中的spark.student表發(fā)生了什么

39、變化mysql select * from student;+ + + + +| id | name | gender | age |+ + + + +| 1 | Xueqian | F | 23 | 2 | Weiliang | M | 24 | 3 | Rongcheng | M | 26 | 4 | Guanhua | M | 27 |+ + + + + 4 rows in set (0.00 sec)Hive簡介和安裝 Hive是一個(gè)構(gòu)建于Hadoop頂層的數(shù)據(jù)倉庫工具 支持大規(guī)模數(shù)據(jù)存儲、分析，具有良好的可擴(kuò)展性 某種程度上可以看作是用戶編程接口，本身不存儲和處理數(shù)據(jù) 依賴分布式文件

40、系統(tǒng)HDFS存儲數(shù)據(jù) 依賴分布式并行計(jì)算模型MapReduce處理數(shù)據(jù) 定義了簡單的類似SQL 的查詢語言HiveQL 用戶可以通過編寫的HiveQL語句運(yùn)行MapReduce任務(wù) 可以很容易把原來構(gòu)建在關(guān)系數(shù)據(jù)庫上的數(shù)據(jù)倉庫應(yīng)用程序移植到Hadoop平臺上 是一個(gè)可以提供有效、合理、直觀組織和使用數(shù)據(jù)的分析工具Hive簡介和安裝 Hive依賴于HDFS 存儲數(shù)據(jù) Hive依賴于MapReduce 處理數(shù)據(jù)Hadoop生態(tài)系統(tǒng)Hive與Hadoop生態(tài)系統(tǒng)中其他組件的關(guān)系HDFSMapReduceHBaseHivePigBI報(bào)表ETL工具Hive簡介和安裝輸入ParserSemantic An

41、alyzerLogical Plan GeneratorLogical OptimizerPhysical Plan GeneratorPhysical Optimizer輸出 選擇最佳的優(yōu)化查詢策略 將邏輯計(jì)劃轉(zhuǎn)成物理計(jì)劃 重寫邏輯查詢計(jì)劃 將查詢塊轉(zhuǎn)換成邏輯查詢計(jì)劃 將抽象語法樹轉(zhuǎn)換成查詢塊 將SQL轉(zhuǎn)換成抽象語法樹讓Spark包含Hive支持 為了讓Spark能夠訪問Hive，必須為Spark添加Hive支持 Spark官方提供的預(yù)編譯版本，通常是不包含Hive支持的，需要采用源碼編譯，編譯得到一個(gè)包含Hive支持的Spark版本（1）測試一下自己電腦上已經(jīng)安裝的Spark版本是否支持H

42、ive啟動進(jìn)入了spark-shell，如果不支持Hive，會顯示如下信息：如果你當(dāng)前電腦上的Spark版本包含Hive支持，那么應(yīng)該顯示下面的正確信息：讓Spark包含Hive支持（2）采用源碼編譯方法得到支持Hive的Spark版本到Spark官網(wǎng)下載源碼/downloads.html讓Spark包含Hive支持解壓文件在編譯時(shí)，需要給出電腦上之前已經(jīng)安裝好的Hadoop的版本 $ hadoop version 運(yùn)行編譯命令，對Spark源碼進(jìn)行編譯編譯成功后會得到文件名“spark-2.1.0-bin-h27hive.tgz”，這個(gè)就是包含Hi

43、ve支持的Spark安裝文件$ cd /home/hadoop/spark-2.1.0$ ./dev/make-distribution.sh tgz name h27hive -Pyarn -Phadoop-2.7 - Dhadoop.version=2.7.1 -Phive -Phive-thriftserver -DskipTests$ cd /home/hadoop/下載 /spark-2.1.0.tgz就在這個(gè)目錄下面$ ls #可以看到剛才下載的spark-2.1.0.tgz文件$ sudo tar -zxf ./spark-2.1.0.tgz -C /home/hadoop/$

44、cd /home/hadoop$ ls #這時(shí)可以看到解壓得到的文件夾spark-2.1.0 讓Spark包含Hive支持（3）安裝支持Hive的Spark版本Spark的安裝詳細(xì)過程，請參考實(shí)驗(yàn)手冊啟動進(jìn)入了spark-shell，由于已經(jīng)可以支持Hive，會顯示如下信息：假設(shè)已經(jīng)完成了Hive的安裝，并且使用的是MySQL數(shù)據(jù)庫來存放Hive的元數(shù)據(jù)需要借助于MySQL保存Hive的元數(shù)據(jù)，首先啟動MySQL數(shù)據(jù)庫： $ service mysql start 在Hive中創(chuàng)建數(shù)據(jù)庫和表由于Hive是基于Hadoop的數(shù)據(jù)倉庫，使用HiveQL語言撰寫的查詢語句，最終都會被H

45、ive自動解析成MapReduce任務(wù)由Hadoop去具體執(zhí)行，因此，需要啟動Hadoop，然后再啟動Hive啟動Hadoop：Hadoop啟動成功以后，可以再啟動Hive：進(jìn)入Hive，新建一個(gè)數(shù)據(jù)庫sparktest，并在這個(gè)數(shù)據(jù)庫下面創(chuàng)建一個(gè)表student，并錄入兩條數(shù)據(jù)hive create database if not exists sparktest;/創(chuàng)建數(shù)據(jù)庫sparktest hive show databases; /顯示一下是否創(chuàng)建出了sparktest數(shù)據(jù)庫/下面在數(shù)據(jù)庫sparktest中創(chuàng)建一個(gè)表studenthive create table if not e

46、xists sparktest.student( id int, name string, gender string, age int);hive use sparktest; /切換到sparktesthive show tables; /顯示sparktest數(shù)據(jù)庫下面有哪些表hive insert into student values(1,Xueqian,F,23); /插入一條記錄hive insert into student values(2,Weiliang,M,24); /再插入一條記錄hive select * from student; /顯示student表中的記錄$

47、 cd /usr/local/hive$ ./bin/hive$ cd /usr/local/hadoop$ ./sbin/start-all.sh連接Hive讀寫數(shù)據(jù)需要修改“/usr/local/sparkwithhive/conf/spark-env.sh”這個(gè)配置文件：export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 export CLASSPATH=$CLASSPATH:/usr/local/hi

48、ve/lib export SCALA_HOME=/usr/local/scalaexport HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoopexport HIVE_CONF_DIR=/usr/local/hive/confexport SPARK_CLASSPATH=$SPARK_CLASSPATH:/usr/local/hive/lib/mysql- connector-java-5.1.40-bin.jar連接Hive讀寫數(shù)據(jù)請?jiān)趕park-shell（包含Hive支持）中執(zhí)行以下命令從Hive中讀取數(shù)據(jù)：Scala import org.apache.spark.sql.RowScala import org.a