理解Spark SQL(二）—— SQLContext和HiveContext

使用Spark SQL，除了使用之前介紹的方法，實際上還可以使用SQLContext或者HiveContext通過編程的方式實作，前者支持SQL語法決議器（SQL-92語法），后者支持SQL語法決議器和HiveSQL語法決議器，默認為HiveSQL語法決議器，用戶可以通過配置切換成SQL語法決議器來運行HiveQL不支持的語法，如：select 1，實際上HiveContext是SQLContext的子類，因此在HiveContext運行程序中除了override的函式和變數，可以使用和SQLContext一樣的函式和變數，

因為spark-shell工具實際就是運行的scala程式片段，為了方便，下面采用spark-shell進行演示，

首先來看SQLContext，因為是標準SQL，可以不依賴于Hive的metastore，比如下面的例子（沒有啟動hive metastore）：

[root@BruceCentOS4 ~]# $SPARK_HOME/bin/spark-shell --master yarn --conf spark.sql.catalogImplementation=in-memory

 scala> case class offices(office:Int,city:String,region:String,mgr:Int,target:Double,sales:Double)
defined class offices

scala> val rddOffices=sc.textFile("/user/hive/warehouse/orderdb.db/offices/offices.txt").map(_.split("\t")).map(p=>offices(p(0).trim.toInt,p(1),p(2),p(3).trim.toInt,p(4).trim.toDouble,p(5).trim.toDouble))
rddOffices: org.apache.spark.rdd.RDD[offices] = MapPartitionsRDD[3] at map at <console>:26

scala> val officesDataFrame = spark.createDataFrame(rddOffices)
officesDataFrame: org.apache.spark.sql.DataFrame = [office: int, city: string ... 4 more fields]

scala> officesDataFrame.createOrReplaceTempView("offices")

scala> spark.sql("select city from offices where region='Eastern'").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

scala>

 執行上面的命令后，實際上在yarn集群中啟動了一個yarn client模式的Spark Application，然后在scala>提示符后輸入的陳述句會生成RDD的transformation，最后一條命令中的collect會生成RDD的action，即會觸發Job的提交和程式的執行，

命令列中之所以加上--conf spark.sql.catalogImplementation=in-memory選項，是因為spark-shell中的默認啟動的SparkSession物件spark是默認支持Hive的，不帶這個選項啟動的話，程式就會去連接hive metastore，因為這里并沒有啟動hive metastore，因此程式在執行createDataFrame函式時會報錯，

程式中的第一行是1個case class陳述句，這里是定義后面的資料檔案的模式的（定義模式除了這個方法，其實還有另外一種方法，后面再介紹），第二行從hdfs中讀取一個文本檔案，并工通過map映射到了模式上面，第三行基于第二行的RDD生成DataFrame，第四行基于第三行的DataFrame注冊了一個邏輯上的臨時表，最后一行就可以通過SparkSession的sql函式來執行sql陳述句了，

實際上，SQLContext是Spark 1.x中的SQL入口，在Spark 2.x中，使用SparkSession作為SQL的入口，但是為了向后兼容，Spark 2.x仍然支持SQLContext來操作SQL，不過會提示deprecated，所以上面的例子是采用Spark 2.x中的寫法，

實際上還有另外一種方法來操作SQL，針對同樣的資料，例如：

scala> import org.apache.spark.sql._
import org.apache.spark.sql._

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> val schema = new StructType(Array(StructField("office", IntegerType, false), StructField("city", StringType, false), StructField("region", StringType, false), StructField("mgr", IntegerType, true), StructField("target", DoubleType, true), StructField("sales", DoubleType, false)))
schema: org.apache.spark.sql.types.StructType = StructType(StructField(office,IntegerType,false), StructField(city,StringType,false), StructField(region,StringType,false), StructField(mgr,IntegerType,true), StructField(target,DoubleType,true), StructField(sales,DoubleType,false))

scala> val rowRDD = sc.textFile("/user/hive/warehouse/orderdb.db/offices/offices.txt").map(_.split("\t")).map(p => Row(p(0).trim.toInt,p(1),p(2),p(3).trim.toInt,p(4).trim.toDouble,p(5).trim.toDouble))
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[3] at map at <console>:30

scala> val dataFrame = spark.createDataFrame(rowRDD, schema)
dataFrame: org.apache.spark.sql.DataFrame = [office: int, city: string ... 4 more fields]

scala> dataFrame.createOrReplaceTempView("offices")

scala> spark.sql("select city from offices where region='Eastern'").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

這個例子與之前的例子有一些不同，主要的地方有3個：

1. 之前的例子是采用case class定義模式，Spark采用反射來推斷Schema；而這個例子采用StructType型別的物件來定義模式，它接收一個陣列，陣列成員是StructField物件，代表一個欄位的定義，每個欄位的定義由欄位名稱、欄位型別和是否允許為空組成；

2. 對于代表資料的RDD，之前的例子是直接用case class定義的型別來分割欄位，而這個例子是用的Row型別；

3. 在使用createDataFrame函式生成DataFrame時，該函式的引數不一樣，之前的例子只要傳入RDD物件即可（物件中隱含了模式），而這個例子需要同時傳入RDD和定義的schema；

實際編程中建議采用第二種方法，因為其更加靈活，schema資訊可以不必是寫死的，而是可以在程式運行的程序中生成，

下面接著來看HiveContext的用法，使用HiveContext之前需要確保：

• 使用的Spark是支持Hive的；
• Hive的組態檔hive-site.xml已經在Spark的conf目錄下；
• hive metastore已經啟動；

舉例說明：

首先啟動hive metastore：

[root@BruceCentOS ~]# nohup hive --service metastore &

然后仍然通過spark-shell來舉例說明，啟動spark-shell，如下所示：

[root@BruceCentOS4 ~]# $SPARK_HOME/bin/spark-shell --master yarn

scala> spark.sql("show databases").collect.foreach(println)
[default]
[orderdb]

scala> spark.sql("use orderdb")
res2: org.apache.spark.sql.DataFrame = []

scala> spark.sql("show tables").collect.foreach(println)
[orderdb,customers,false]
[orderdb,offices,false]
[orderdb,orders,false]
[orderdb,products,false]
[orderdb,salesreps,false]

scala> spark.sql("select city from offices where region='Eastern'").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

scala>

可以看到這次啟動spark-shell沒有帶上最后那個選項，這是因為這里我們打算用HiveContext來操作Hive中的資料，需要支持Hive，前面說過spark-shell是默認開啟了Hive支持的，同SQLContext類似，Spark 2.x中也不需要再用HiveContext物件來操作SQL了，直接用SparkSession物件來操作就好了，可以看到這里可以直接操作表，不用再定義schema，這是因為schema是由外部的hive metastore定義的，spark通過連接到hive metastore來讀取表的schema資訊，因此這里能直接操作SQL，

另外，除了上面的使用SQLContext操作普通檔案（需要額外定義模式）和使用HiveContext操作Hive表資料（需要開啟hive metastore）之外，SQLContext還能操作JSON、PARQUET等檔案，由于這兩種資料檔案自己帶了模式資訊，因此可以直接基于檔案創建DataFrame，例如：

scala> val df = spark.read.json("file:///opt/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]                

scala> df.createOrReplaceTempView("people")

scala> spark.sql("select name,age from people where age>19").map(t=>"Name :" + t(0) + ", Age: " + t(1)).collect.foreach(println)
Name :Andy, Age: 30    

最后來看下DataFrame的另一種叫做DSL(Domain Specific Language)的用法，

scala> val df = spark.read.json("file:///opt/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]                

scala> df.show()
+----+-------+                                                                  
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+


scala> df.select("name").show()
+-------+                                                                       
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+


scala> df.select(df("name"), df("age") + 1).show()
+-------+---------+                                                             
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+


scala> df.filter(df("age") > 21).show()
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+


scala> df.groupBy("age").count().show()
+----+-----+                                                                    
| age|count|
+----+-----+
|  19|    1|
|null|    1|
|  30|    1|
+----+-----+


scala>

以上是對Spark SQL的SQLContext和HiveContext基本用法的一些總結，都是采用spark-shell工具舉的例子，實際上由于spark-shell是運行scala程式片段的工具，上述例子完全可以改成獨立的應用程式，我將在下一篇博文當中嘗試使用Scala、Java和Python來撰寫獨立的程式來操作上面的示例hive資料庫orderdb，可以適當使用一些較為復雜的SQL來統計分析資料，

標籤：大數據

上一篇：spark利用yarn提交任務報:YARN application has exited unexpectedly with state UNDEFINED

下一篇：第四天MySQL