從hive到Spark SQL

01-27

＊SQLContext／HiveContext／SparkSession的使用

註：SQLContext／HiveContext只有在1.x裡面有，2.x已經沒有了，但是為了更好的理解SparkSQL，我們仍然需要給予1.x版本里的這個功能一些關注。在2.x裡面就是我們看到的SparkSession了。

＊sparks-shell／spark－sql的使用

＊thriftserver／beeline的使用

＊jdbc方式編程

No.1 SQLContext／HiveContext／SparkSession

官網地址：1.6.1版本

我們在IDEA上面進行編程，首先是pom.xml的配置：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">n <modelVersion>4.0.0</modelVersion>n <groupId>com.data.spark</groupId>n <artifactId>sql</artifactId>n <version>1.0</version>n <inceptionYear>2008</inceptionYear>n <properties>n <scala.version>2.11.8</scala.version>n <spark.version>2.2.0</spark.version>n </properties>nn <repositories>n <repository>n <id>scala-tools.org</id>n <name>Scala-Tools Maven2 Repository</name>n <url>http://scala-tools.org/repo-releases</url>n </repository>n </repositories>nn <pluginRepositories>n <pluginRepository>n <id>scala-tools.org</id>n <name>Scala-Tools Maven2 Repository</name>n <url>http://scala-tools.org/repo-releases</url>n </pluginRepository>n </pluginRepositories>nn <dependencies>n n <dependency>n <groupId>org.scala-lang</groupId>n <artifactId>scala-library</artifactId>n <version>${scala.version}</version>n </dependency>n <dependency>n <groupId>junit</groupId>n <artifactId>junit</artifactId>n <version>4.4</version>n <scope>test</scope>n </dependency>n <dependency>n <groupId>org.specs</groupId>n <artifactId>specs</artifactId>n <version>1.2.5</version>n <scope>test</scope>n </dependency>nn n <dependency>n <groupId>org.apache.spark</groupId>n <artifactId>spark-sql_2.11</artifactId>n <version>${spark.version}</version>n </dependency>nn n <dependency>n <groupId>org.apache.spark</groupId>n <artifactId>spark-hive_2.11</artifactId>n <version>${spark.version}</version>n </dependency>nn </dependencies>nn <build>n <sourceDirectory>src/main/scala</sourceDirectory>n <testSourceDirectory>src/test/scala</testSourceDirectory>n <plugins>n <plugin>n <groupId>org.scala-tools</groupId>n <artifactId>maven-scala-plugin</artifactId>n <executions>n <execution>n <goals>n <goal>compile</goal>n <goal>testCompile</goal>n </goals>n </execution>n </executions>n <configuration>n <scalaVersion>${scala.version}</scalaVersion>n <args>n <arg>-target:jvm-1.5</arg>n </args>n </configuration>n </plugin>n <plugin>n <groupId>org.apache.maven.plugins</groupId>n <artifactId>maven-eclipse-plugin</artifactId>n <configuration>n <downloadSources>true</downloadSources>n <buildcommands>n <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>n </buildcommands>n <additionalProjectnatures>n <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>n </additionalProjectnatures>n <classpathContainers>n <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>n <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>n </classpathContainers>n </configuration>n </plugin>n </plugins>n </build>n <reporting>n <plugins>n <plugin>n <groupId>org.scala-tools</groupId>n <artifactId>maven-scala-plugin</artifactId>n <configuration>n <scalaVersion>${scala.version}</scalaVersion>n </configuration>n </plugin>n </plugins>n </reporting>n</project>n

然後就是我們的代碼了：

package com.data.sparknnimport org.apache.spark.{SparkConf, SparkContext}nimport org.apache.spark.sql.SQLContextnn/**n * SQLContext的使用n */nnobject SparkContext {nn def main(args: Array[String]): Unit = {nn val path = args(0)nn //1.創建相應的Contextn val sparkConf = new SparkConf()nn //在測試或者生產中，AppName和Master通過腳本進行指定n //sparkConf.setAppName("SQLContextApp").setMaster("local[2]")nn val sc = new SparkContext(sparkConf)n val sqLContext = new SQLContext(sc) //這裡顯示過時了，這主要是演示1.x的使用nn //2.處理一個json文件n val people = sqLContext.read.format("json").load(path)n people.printSchema()n people.show()n //3.關閉資源n sc.stop()n }n}n

這裡我們//sparkConf.setAppName("SQLContextApp").setMaster("local[2]")這個是注釋掉的，因為我們不再idea中運行，我們直接用mvn命令打包成jar，然後在環境中運行。

首先終端進入project目錄，然後執行：

mvn clean package -DskipTestsn

我們看到他執行成功了，然後我們運行他，首先我們建一個shell文件夾，裡面放.sh的執行文件，當然這些需要我們自己創建，因為我們把命令放到這些.sh文件，我們直接執行這些.sh文件就可以了，.sh文件建好之後需要給予他們執行許可權，賦許可權命令：

chmod u+x 文件名.shn

然後是執行的命令：

spark-submit n --class com.data.spark.SparkContext n --master local[2] n /Users/yinchuchu/Desktop/SparkSQLProject/target/sql-1.0.jar n /Users/yinchuchu/Downloads/software/hadoop/spark-2.2.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.jsonn

以下就是操作好了的文件了。

最後一行最長的那個是測試的json文件，在spark里買的example裡面找，就按照上面spark安裝目錄下的路徑就能找到。

然後我們執行看看。

執行成功。

2、Spark1.x中Spark SQL的入口點：HiveContext

代碼：

package com.data.sparknnimport org.apache.spark.sql.hive.HiveContextnimport org.apache.spark.{SparkConf, SparkContext}nn/**n * HiveContext的使用n * 使用時需要通過--jars 把mysql的驅動傳遞到classpathn */nobject HiveContext {nn def main(args: Array[String]) {n //1)創建相應的Contextn val sparkConf = new SparkConf()nn //在測試或者生產中，AppName和Master我們是通過腳本進行指定n sparkConf.setAppName("HiveContextApp").setMaster("local[2]")nn val sc = new SparkContext(sparkConf)n val hiveContext = new HiveContext(sc)nn //2)相關的處理:n hiveContext.table("emp").shownn //3)關閉資源n sc.stop()n }n}n

同樣的和sqlcontext一樣，建.sh文件，賦許可權，執行。

首先是執行命令：

spark-submit n --class com.data.spark.HiveContext n --master local[2] n --jars /Users/yinchuchu/Downloads/software/hadoop/hive-1.1.0-cdh5.7.0/lib/mysql-connector-java-5.1.27-bin.jar n /Users/yinchuchu/Desktop/SparkSQLProject/target/sql-1.0.jar n

這裡我們不再是example里的people.json了，而是直接處理hive裡面的emp這份員工表，這個員工表之前的文章裡面寫到過，之前操作留下的，可以看著之前的數據倉庫Hive的使用。

然後命令中要用jars帶上mysql的驅動才能執行，（mysql驅動包下載：https://pan.baidu.com/s/1o8C5cF4 密碼: 7sxb)

然後執行.sh文件

出現表格說明運行成功。

3、Spark1.x中Spark SQL的入口點：SparkSession

我們還是看官網，這個時候就要去看最新的文檔了，看2.2.0版本的。

好，那我們寫代碼測試一下。

package com.data.sparknnimport org.apache.spark.sql.SparkSessionnn/*n* SparkSession的使用n* */nobject SparkSessionApp {n def main(args: Array[String]): Unit = {n val spark = SparkSession.builder().appName("SparkSession").master("local[2]").getOrCreate()nn val people = spark.read.json("file:///Users/yinchuchu/Downloads/software/hadoop/spark-2.2.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.json")n people.show()nn spark.stop()n }n}n

然後我們看運行結果。

正常運行，解析出了這個json文件。

No.2 spark-shell／spark－sql的使用

我們之前已經在hive裡面創建了兩張表，這兩張表我們如何能用spark-shell或者spark-sql訪問呢？我們先去控制台看看。當然首先要在 Hadoop的sbin目錄下啟動hadoop。再把mysql.server給啟動起來。然後進到hive裡面看看兩張表格。

好，看到了這兩張表格了。然後我們再通過spark-shell來訪問。

我們現在什麼都沒有配置，什麼都還沒有，如果要訪問 hive，我們有兩個條件，一個是需要hive-site.xml 的，這個在hive安裝目錄下的conf裡面找。

我們看這個裡面的藍色標記，因為這個hive的元數據是存儲在mysql裡面的，所以我們要指定連接的url，指定連接drivername，指定連接用戶名和連接密碼。你 spark要訪問元數據裡面的信息，我們首先第一步需要把這個hive-site.xml拷貝到spark下面的conf目錄中的。我們來拷貝一下。

好了，已經拷貝進來了。還有一個步驟，就是我們需要mysql的驅動包，因為元數據shizaimysql裡面的，你要訪問數據，你肯定得知道元數據，所以你要把mysql驅動包帶上，怎麼帶呢？很簡單，在命令後面加上驅動包的路徑就可以了，比如我的命令是這樣的：

spark-shell --master local[2] n--jars /Users/yinchuchu/Downloads/software/hadoop/hive-1.1.0-cdh5.7.0/lib/mysql-connector-java-5.1.27-bin.jarn

後面加上--jars mysql驅動包存放路徑就可以執行了。

看到沒有！我們成功拿到數據表了！如果這裡你這樣操作你還是沒有拿到數據，那麼你可以這樣驗證一下，你就在spark終端import org.apache.spark.sql.hive.HiveContext

如果你執行後是這樣的：

scala> import org.apache.spark.sql.hive.HiveContextn<console>:25: error: object hive is not a member of package org.apache.spark.sqln import org.apache.spark.sql.hive.HiveContextn ^n

那不好意思，很可能你源碼編譯有問題，嘗試這重新編譯吧！雖然時間稍微長了點，但是很有可能你在網上找半天都解決不了，那你不如一開始就重新編譯，說不定還快些呢。

編譯後在測試，如果是這樣的說明正常了：

scala> import org.apache.spark.sql.hive.HiveContextnimport org.apache.spark.sql.hive.HiveContextn

好，既然成功訪問了，我們再來看看能不能查看數據內容。

看到沒，我們成功訪問了數據表中的內容！

我們再給emp和dept這兩個表做一個join來看看。

好了，成功了。但是我們每次都要使用Scala來操作，能不能直接使用sql呢？只要輸入SQL就能拿到結果，這是可以的，這個時候就需要用到另一個腳本了，叫做spark.sql

我們進到spark安裝目錄下面bin裡面就能看到這個文件了。

執行方法和spark_shell是一樣的。我們執行看看。

啟動過程的截圖，我們看到，在4041這個埠監控我們的作業。

我們進去這個埠看看。

確實是這樣，現在還是空的，我們在spark-sql執行一個sql看看。

select * from emp;n

我們看到終端正確輸出數據了，我們再來看看監控頁面上面。

當然我們在spark-sql裡面也是可以寫join的，sql語句你怎麼寫的，這裡就怎麼用的。

select * from emp e join dept d on e.deptno=d.deptno;n

也成功輸出數據了。再到監控頁面上看看。

到這裡，我們就已經實現了spark sql操作hive數據倉庫，那麼我們在這裡測試一個可以幫助我們理解spark架構的一個操作。

首先我們在spark-sql創建一張表。

create table t(key string, value string);n

現在沒有數據，不用管，我們現在來看一個執行計劃。

explain extended select a.key*(2+3), b.value from t a join t b on a.key = b.key and a.key > 3;n

這是隨意寫的，主要看看執行是什麼樣子的。

我們好像看到了一些熟悉的東西，我們把內容單獨拷貝出來。

== Parsed Logical Plan ==nProject [unresolvedalias((a.key * (2 + 3)), None), b.value]n+- Join Inner, ((a.key = b.key) && (a.key > 3))n :- SubqueryAlias an : +- UnresolvedRelation `t`n +- SubqueryAlias bn +- UnresolvedRelation `t`nn== Analyzed Logical Plan ==n(CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE)): double, value: stringnProject [(cast(key#56 as double) * cast((2 + 3) as double)) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- Join Inner, ((key#56 = key#58) && (cast(key#56 as int) > 3))n :- SubqueryAlias an : +- SubqueryAlias tn : +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- SubqueryAlias bn +- SubqueryAlias tn +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]nn== Optimized Logical Plan ==nProject [(cast(key#56 as double) * 5.0) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- Join Inner, (key#56 = key#58)n :- Project [key#56]n : +- Filter (isnotnull(key#56) && (cast(key#56 as int) > 3))n : +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- Filter ((cast(key#58 as int) > 3) && isnotnull(key#58))n +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]nn== Physical Plan ==n*Project [(cast(key#56 as double) * 5.0) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- *SortMergeJoin [key#56], [key#58], Innern :- *Sort [key#56 ASC NULLS FIRST], false, 0n : +- Exchange hashpartitioning(key#56, 200)n : +- *Filter (isnotnull(key#56) && (cast(key#56 as int) > 3))n : +- HiveTableScan [key#56], CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- *Sort [key#58 ASC NULLS FIRST], false, 0n +- Exchange hashpartitioning(key#58, 200)n +- *Filter ((cast(key#58 as int) > 3) && isnotnull(key#58))n +- HiveTableScan [key#58, value#59], CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]n

我們一個一個看。

1）首先，他生成一個邏輯執行計劃，也就是第一段。

肯定是通過抽象語法樹給他解析成邏輯執行計劃，你看這裡面也有一些[unresolvedalias，就是沒有解析全的別名，比如我們的a.key * (2 + 3)，是一個None類型，其實這裡面欄位就是我們select a.key*(2+3), b.value 這一段。它採用了Join Inner的方式，條件是什麼呢？

((a.key = b.key) && (a.key > 3))，這個好理解，是一個and的關係，這一句其實就是on後面的條件，再往下還有UnresolvedRelation，首先有一個t別名是a，還有一個t別名叫b，其實就是自己和自己做了一個join，以上就是邏輯執行計划了。

2）再往下看第二段。

== Analyzed Logical Plan ==n(CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE)): double, value: stringnProject [(cast(key#56 as double) * cast((2 + 3) as double)) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- Join Inner, ((key#56 = key#58) && (cast(key#56 as int) > 3))n :- SubqueryAlias an : +- SubqueryAlias tn : +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- SubqueryAlias bn +- SubqueryAlias tn +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]n

我們對邏輯執行計劃做一個解析，就是這一段Analyzed Logical Plan。

這一段說白了是要與我們的底層metastore打交道的，我們看，首先key本來是string類型，但是我們用a.key * (2 + 3)，那麼他首先把key轉換成一個double類型，就是這一句(cast(key#56 as double)，最終a.key*(2+3)變成一個double類型，然後value還是一個string類型，下面有一個project，這個project就是select裡面的兩個欄位，Project [(cast(key#56 as double) * cast((2 + 3) as double)) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]對應的就是a.key*(2+3), b.value，下面還是一個Join Inner，再往下有一個SubqueryAlias a

就是子查詢別名，就是a表，這個a他現在他已經知道metastore對應的關係，已經知道他是default資料庫裡面的t表，下面的也是t表。實際上這一整段是在上一段基礎之上和metastore做的一些關聯的執行操作，看這一段的名稱就是Analyzed Logical Plan。

3）再看第三段，我們可以對這個執行計划進行優化。那麼到底可以優化哪些呢？我們來看。

== Optimized Logical Plan ==nProject [(cast(key#56 as double) * 5.0) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- Join Inner, (key#56 = key#58)n :- Project [key#56]n : +- Filter (isnotnull(key#56) && (cast(key#56 as int) > 3))n : +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- Filter ((cast(key#58 as int) > 3) && isnotnull(key#58))n +- CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]n

首先這個SQL語句是on a.key = b.key，就是先對這個key相同的做了一個join，然後在加上過濾條件 a.key > 3，其實如果熟悉SQL，就知道，他會先把 a.key > 3這個過濾條件先給他過濾掉，然後在做join，他的數據量是不是會少一些，對吧。先過濾再做join，那麼spark-sql自己就給我們做了這一層優化，我們看這一段，Join Inner下面，首先有一個Filter (isnotnull(key#56) && (cast(key#56 as int) > 3))是不是，這一段處理之後，我們的數據就壓下去一些，先把滿足條件的拿出來，不滿足條件的先過濾掉。優化之後再生成一個物理執行計劃。

4）物理執行計劃Physical Plan。

== Physical Plan ==n*Project [(cast(key#56 as double) * 5.0) AS (CAST(key AS DOUBLE) * CAST((2 + 3) AS DOUBLE))#60, value#59]n+- *SortMergeJoin [key#56], [key#58], Innern :- *Sort [key#56 ASC NULLS FIRST], false, 0n : +- Exchange hashpartitioning(key#56, 200)n : +- *Filter (isnotnull(key#56) && (cast(key#56 as int) > 3))n : +- HiveTableScan [key#56], CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#56, value#57]n +- *Sort [key#58 ASC NULLS FIRST], false, 0n +- Exchange hashpartitioning(key#58, 200)n +- *Filter ((cast(key#58 as int) > 3) && isnotnull(key#58))n +- HiveTableScan [key#58, value#59], CatalogRelation `default`.`t`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [key#58, value#59]n

這個物理執行計劃，你看他採用的排序SortMergeJoin，這一段了解一下就好。

主要的是整個過程。

spark-shell／spark－sql的使用的總結：

hive-site.xml配置文件拷貝到spark的conf下
--jars傳遞mysql驅動包

No.3 thriftserver／beeline的使用

首先進入spark安裝目錄，進入sbin，找到start-thriftserver.sh。然後直接運行。

運行命令其實和啟動spark-shell一樣的：

./start-thriftserver.sh --master local[2] --jars n/Users/yinchuchu/Downloads/software/hadoop/hive-1.1.0-cdh5.7.0/lib/mysql-connector-java-5.1.27-bin.jarn

注意運行後，要cat一下log查看有沒有錯誤，你運行完了，回提示你log的位置，如圖藍色標記部分。

一般有錯誤可能你忘記帶上jars後面的mysql驅動了，要注意。然後可以jps -m查看一下有沒有parkSubmit --master local[2] --class org.apache.spark.sql.hive.thriftserver.HiveThriftServer2 --name Thrift JDBC/ODBC Server --jars /Users/yinchuchu/Downloads/software/hadoop/hive-1.1.0-cdh5.7.0/lib/mysql-connector-java-5.1.27-bin.jar spark-internal這個進程，有的話就是成功啟動了。

然後我們去監控頁面上看一下。

也是有spark頁面的。現在server啟動了，我們要通過beeline，也就是客戶端來連接對不對，怎麼連接呢？看官網！

藍色標記那一段jdbc:hive2://localhost:10000

1) 啟動thriftserver: 默認埠是10000 ，可以修改

2) 啟動beeline，在spark安裝目錄下面的bin目錄找到beeline命令

beeline -u jdbc:hive2://localhost:10000 -n localhostn

我們看到連接成功並且可以獲取到hive裡面的表格，我們再看看數據，看數據之前注意要把hadoop啟動起來，不然獲取數據失敗。

好，數據我們也看到了，我們再做一個join命令看看。

成功join了有沒有，我們再去監控頁面上看看。

看到沒有，反饋信息十分詳細，還有這個sql的解析過程都有。

好了，現在我們是通過一個客戶端連接上來的，我們能不能在啟動一個客戶端呢，

肯定可以的，直接新建一個終端執行命令運行就可以了。

發現沒有，變成了2個session。還有我們看到啟動命令默認埠是10000，這是可以修改的，如何改呢？看官網呀。

這段命令就可以，我們拿來改一下。

./start-thriftserver.sh n --hiveconf hive.server2.thrift.port=14000 n --master local[2] n --jars /Users/yinchuchu/Downloads/software/hadoop/hive-1.1.0-cdh5.7.0/lib/mysql-connector-java-5.1.27-bin.jarn

我們重新啟動thriftserver。首先在spark的sbin目錄下停止thriftserver服務。

然後，再把上述命令執行一下。

好，沒問題了。我們再次啟動beeline。注意這個時候啟動就是14000埠了。

成功啟動，我們在做一個sql看看。

ok!全部出來了！

thriftserver和普通的spark-shell/spark-sql有什麼區別呢？

1）spark-shell、spark-sql都是一個spark application；

2）thriftserver，不管你啟動多少個客戶端(beeline/code)，永遠都是一個spark application

解決了一個數據共享的問題，多個客戶端可以共享數據；也就是說啟動server申請一次資源就可以了，後面不管你啟動多少客戶端都不需要申請資源。

No.4 jdbc方式編程訪問

＊maven添加依賴：org.spark-project.hive#hive-jdbc

＊開發代碼訪問thriftserver

好了，我們進入IDEA進行想應的開發

package com.data.sparknnimport java.sql.DriverManagernn/*n *通過jdbc的方式訪問n */nobject SparkSQLThriftServerApp {n def main(args: Array[String]): Unit = {n Class.forName("org.apache.hive.jdbc.HiveDriver")nn val conn = DriverManager.getConnection("jdbc:hive2://localhost:14000", "localhost", "")n val pstmt = conn.prepareStatement("select empno, ename, sal from emp")n val rs = pstmt.executeQuery()n while (rs.next()){n println("empno:" + rs.getInt("empno")+n ", ename:" + rs.getString("ename") +n ", sal:" + rs.getDouble("sal"))n }n rs.close()n pstmt.close()n conn.close()n }nn}n

看到沒有，我們通過jdbc的方式訪問數據成功了。

這裡有一點要注意，用jdbc之前一定要先把thriftserver打開，不然是不行的。

到此，從hive到Spark SQ就結束了！下面該寫寫DataFrame和DataSet了。。。