【導讀：數據是二十一世紀的石油，蘊含巨大價值，這是·情報通·大數據技術系列第[82]篇文章，歡迎閱讀和收藏】

1 背景介紹

SparkSQL 引入了一種新的 RDD —— SchemaRDD ， SchemaRDD 由行對象（ Row ）以及描述行對象中每列數據類型的 Schema 組成； SchemaRDD 很像傳統數據庫中的表。 SchemaRDD 可以通過 RDD 、 Parquet 文件、 JSON 文件、或者通過使用 hiveql 查詢 hive 數據來建立。 SchemaRDD 除了可以和 RDD 一樣操作外，還可以通過 registerTempTable 註冊成臨時表，然後通過 SQL 語句進行操作。

值得注意的是：

l Spark1.1 使用 registerTempTable 代替 1.0 版本的 registerAsTable

l Spark1.1 在 hiveContext 中， hql() 將被棄用， sql() 將代替 hql() 來提交查詢語句，統一了接口。

l 使用 registerTempTable 註冊表是一個臨時表，生命週期只在所定義的 sqlContext 或 hiveContext 實例之中。換而言之，在一個 sqlContext （或 hiveContext ）中 registerTempTable 的表不能在另一個 sqlContext （或 hiveContext ）中使用。

l Spark1.1 提供了語法解析器選項 spark.sql.dialect ，就目前而言， Spark1.1 提供了兩種語法解析器： sql 語法解析器和 hiveql 語法解析器。

l sqlContext 現在只支持 sql 語法解析器（ SQL-92 語法）

l hiveContext 現在支持 sql 語法解析器和 hivesql 語法解析器，默認為 hivesql 語法解析器，用戶可以通過配置切換成 sql 語法解析器，來運行 hiveql 不支持的語法，如 select 1 。

切換可以通過下列方式完成：

l 在 sqlContexet 中使用 setconf 配置 spark.sql.dialect

l 在 hiveContexet 中使用 setconf 配置 spark.sql.dialect

l 在 sql 命令中使用 set spark.sql.dialect=value

SparkSQL1.1 對數據的查詢分成了 2 個分支： sqlContext 和 hiveContext 。至於兩者之間的關係， hiveSQL 繼承了 sqlContext ，所以擁有 sqlontext 的特性之外，還擁有自身的特性（最大的特性就是支持 hive ）。

2 sqlContext

2.1 使用 Case Class 定義 RDD

對於 Case Class 方式，首先要定義 Case Class ，在 RDD 的 Transform 過程中使用 Case Class 可以隱式轉化成 SchemaRDD ，然後再使用 registerTempTable 註冊成表。註冊成表後就可以在 sqlContext 對錶進行操作，如 select 、 insert 、 join 等。注意， case class 可以是嵌套的，也可以使用類似 Sequences 或 Arrays 之類複雜的數據類型。

應用案例：

scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)

scala> import sqlContext.implicits._

scala> case class Person(name:String,age:Int)

scala> val people=sc.textFile("hdfs://cloud25:9000/data/people.txt").map(_.split(",")).map(p=>Person(p(0),p(1).trim.toInt)).toDF()

scala> teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

2.2 Parquest 應用案例

sqlContext 可以讀取 parquet 文件，由於 parquet 文件中保留了 schema 的信息，所以不需要使用 case class 來隱式轉換。 sqlContext 讀入 parquet 文件後直接轉換成 SchemaRDD ，也可以將 SchemaRDD 保存成 parquet 文件格式。

應用案例：

scala> import sqlContext.implicits._

scala> sqlContext.setConf("spark.sql.parquet.binaryAsString", "true") // 解決文件中 parquet 中 binary 字段的問題

scala> val wikiData = sqlContext.parquetFile("hdfs://cloud25:9000/data/wiki_parquet").toDF()

scala> wikiData.count()

scala> wikiData.registerTempTable("wikidata")

scala> val countResult = sqlContext.sql("SELECT COUNT(*) FROM wikiData").collect()

scala> val queryResult= sqlContext.sql("SELECT username, COUNT(*) AS cnt FROM wikiData WHERE username <> '' GROUP BY username ORDER BY cnt DESC LIMIT 10")

scala> queryResult.collect().foreach(println)

2.3 Join 應用案例

sqlContext 可以從多個種類的 SchemaRDD 中執行 join 操作

應用案例：

scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)

scala> import sqlContext.implicits._

scala> case class Person(name:String,age:Int)

scala> val people=sc.textFile("hdfs://cloud25:9000/data/people.txt").map(_.split(",")).map(p=>Person(p(0),p(1).trim.toInt)).toDF()

scala> people.saveAsParquetFile("hdfs://cloud25:9000/data/people.parquet")

scala> val parquetFile = sqlContext.parquetFile("hdfs://cloud25:9000/data/people.parquet")

scala> people.registerTempTable("people")

scala> parquetFile.registerTempTable("parquetFile")

scala> val jointbls = sqlContext.sql("SELECT people.name FROM people join parquetFile where people.name=parquetFile.name")

scala> jointbls.collect.foreach(println)

3 hiveContext

SparkSQL 1.4.1 版本，可以動過兩種方式與 hive 集成，第一種是使用已經部署好的 metastore ，如果使用這種方式需要在 spark 中添加 hive 的相關配置，第二種 : 如果沒有配置相關的信息， hiveContext 會在當前目錄自己創建 metadata_db 以及 warehouse

配置過程

以下配置流程僅僅針對描述中的第一種情況（使用已有 metastore ）

l 後臺啟動 metasore 服務

l 創建 /opt/cloud/spark-1.4.1-bin-hadoop2.6/conf/hive-site.xml 文件，並編輯內容如下：

<configuration>

<property>

<name>hive.metastore.uris/<name>

<value>thrift://cloud25:9083/<value>

<description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore./<description>

啟動 spark-shell:

./spark-shell --master=spark://cloud25:7077 --executor-memory=2g

應用案例：

l 使用已有 metastore 的代碼

scala> val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)

scala> hiveContext.sql("select count(*) from SOGOUQ1 where S_SEQ=1 and C_SEQ=2").collect.foreach(println)

scala> hiveContext.sql("select count(*) from SOGOUQ1 where S_SEQ=1 and C_SEQ=2 and WEBSITE like '%baidu%'").collect.foreach(println)

scala> hiveContext.sql("select WEBSESSION,count(WEBSESSION) as cw from SOGOUQ1 group by WEBSESSION order by cw desc limit 10").collect.foreach(println)

l 獨立維護 metastore 的代碼

scala> val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)

scala> hiveContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")

閱讀更多 情報通 的文章

關鍵字: Spark HDFS Hive