1.1  執行環境說明

1.1.1 硬軟體環境

l  主機作業系統：Windows 64位，雙核4執行緒，主頻2.2G，10G記憶體

l  虛擬軟體：VMware® Workstation 9.0.0 build-812388

l  虛擬機器作業系統：CentOS6.5 64位，單核

l  虛擬機器執行環境：

Ø  JDK：1.7.0_55 64位

Ø  Hadoop：2.2.0（需要編譯為64位）

Ø  Scala：2.10.4

Ø  Spark：1.1.0（需要編譯）

Ø  Hive：0.13.1（原始碼編譯，參見1.2）

1.1.2 叢集網路環境

本次實驗環境只需要hadoop1一臺機器即可，網路環境配置如下：

1.2 編譯Hive

1.2.1 下載Hive原始碼包

這裡選擇下載的版本為hive-0.13.1，這個版本需要到apache的歸檔伺服器下載，下載地址：http://archive.apache.org/dist/hive/hive-0.13.1/，選擇apache-hive-0.13.1-src.tar.gz檔案進行下載：

1.2.2 上傳Hive原始碼包

把下載的hive-0.13.0.tar.gz安裝包，使用SSH Secure File Transfer工具（參見第2課《Spark編譯與部署（上）--基礎環境搭建》1.3.1介紹）上傳到/home/hadoop/upload 目錄下。

1.2.3 解壓縮並移動到編譯目錄

到上傳目錄下，用如下命令解壓縮hive安裝檔案：

$cd /home/hadoop/upload

$tar -zxf apache-hive-0.13.1-src.tar.gz

改名並移動到/app/complied目錄下：

$sudo mv apache-hive-0.13.1-src /app/complied/hive-0.13.1-src

$ll /app/complied

1.2.4 編譯Hive

編譯Hive原始碼的時候，需要從網上下載依賴包，所以整個編譯過程機器必須保證在聯網狀態。編譯執行如下指令碼：

$cd /app/complied/hive-0.13.1-src/

$export MAVEN_OPTS="-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512m"

$mvn -Phadoop-2,dist -Dmaven.test.skip=true clean package

在編譯過程中可能出現速度慢或者中斷，可以再次啟動編譯，編譯程式會在上次的編譯中斷處繼續進行編譯，整個編譯過程耗時與網速緊密相關，網速較快的情況需要1個小時左右（上圖的時間是多次編譯後最後成功的介面）。最終編譯的結果為$HIVE_HOME/packaging/target/apache-hive-0.13.1-bin.tar.gz

通過如下命令檢視最終編譯完成整個目錄大小，可以看到大小為353.6M左右

$du -s /app/complied/hive-0.13.1-src

【注】已經編譯好的Hive包在本系列配套資源/install/6.hive-0.13.1-src.tar.gz，讀者直接使用

1.3 首次執行hive-console

1.3.1 獲取Spark原始碼

由於首次執行hive-console需要在Spark原始碼進行編譯，關於Spark原始碼的獲取可以參考第二課《Spark編譯與部署（下）--Spark編譯安裝》方式進行獲取，連線地址為 http://spark.apache.org/downloads.html，獲取原始碼後把Spark原始碼移動到/app/complied目錄，並命名為spark-1.1.0-hive

1.3.2 配置/etc/profile環境變數

第一步   使用如下命令開啟/etc/profile檔案：

$sudo vi /etc/profile

第二步   設定如下引數：

export HADOOP_HOME=/app/hadoop/hadoop-2.2.0

export HIVE_HOME=/app/complied/hive-0.13.1-src

export HIVE_DEV_HOME=/app/complied/hive-0.13.1-src

第三步   生效配置並驗證

$sudo vi /etc/profile

$echo $HIVE_DEV_HOME

1.3.3 執行sbt進行編譯

執行hive/console不需要啟動Spark，需要進入到Spark根目錄下使用sbt/sbt hive/console進行首次執行編譯，編譯以後下次可以直接啟動。編譯Spark原始碼的時候，需要從網上下載依賴包，所以整個編譯過程機器必須保證在聯網狀態。編譯命令如下：

$cd /app/complied/spark-1.1.0-hive

$sbt/sbt hive/console

編譯時間會很長，在編譯過程中可能出現速度慢或者中斷，可以再次啟動編譯，編譯程式會在上次的編譯中斷處繼續進行編譯，整個編譯過程耗時與網速緊密相關。

通過如下命令檢視最終編譯完成整個目錄大小，可以看到大小為267.9M左右

$du -s /app/complied/spark-1.1.0-hive

【注】已經編譯好的Spark for hive-console包在本系列配套資源/install/6.spark-1.1.0-hive.tar.gz，可直接使用

1.4 使用hive-console

1.4.1 啟動hive-console

進入到spark根目錄下，使用如下命令啟動hive-console

$cd /app/complied/spark-1.1.0-hive

$sbt/sbt hive/console

1.4.2 輔助命令Help和Tab鍵

可以使用:help檢視幫助內容

scala>:help

可以使用tab鍵檢視所有可使用命令、函式

1.4.3 常用操作

首先定義Person類，在該類中定義name、age和state三個列，然後把該類註冊為people表並裝載資料，最後通過查詢到資料存放到query中

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

scala>sparkContext.parallelize(Person("Michael",29,"CA")::Person("Andy",30,"NY")::Person("Justin",19,"CA")::Person("Justin",25,"CA")::Nil).registerTempTable("people")

scala>val query= sql("select * from people")

1.4.3.1 檢視查詢的schema

scala>query.printSchema

scala>query.collect()

1.4.3.2 檢視查詢的整個執行計劃

scala>query.queryExecution

1.4.4 檢視查詢的Unresolved LogicalPlan

scala>query.queryExecution.logical

1.4.4.1 檢視查詢的Analyzed LogicalPlan

scala>query.queryExecution.analyzed

1.4.4.2 檢視優化後的LogicalPlan

scala>query.queryExecution.optimizedPlan

1.4.4.3 檢視物理計劃

scala>query.queryExecution.sparkPlan

1.4.4.4 檢視RDD的轉換過程

scala>query.toDebugString

1.4.5 不同資料來源的執行計劃

上面常用操作裡介紹了源自RDD的資料， SparkSQL也可以源自多個數據源：jsonFile、parquetFile和Hive等。

1.4.5.1 讀取Json格式資料

第一步   Json測試資料

Json檔案支援巢狀表，SparkSQL也可以讀入巢狀表，如下面形式的Json資料，可以使用jsonFile讀入SparkSQL。該檔案可以在配套資源/data/class6中找到，在以下測試中把檔案放到 /home/hadoop/upload/class6 路徑中

{ 

"fullname": "Sean Kelly",    

"org": "SK Consulting",    

"emailaddrs": [    

{"type": "work", "value": "[email protected]"},    

{"type": "home", "pref": 1, "value": "[email protected]"}    

],    

"telephones": [    

{"type": "work", "pref": 1, "value": "+1 214 555 1212"},    

{"type": "fax", "value": "+1 214 555 1213"},    

{"type": "mobile", "value": "+1 214 555 1214"}    

],    

"addresses": [    

{"type": "work", "format": "us",    

"value": "1234 Main StnSpringfield, TX 78080-1216"},    

{"type": "home", "format": "us",    

"value": "5678 Main StnSpringfield, TX 78080-1316"}    

],    

"urls": [    

{"type": "work", "value": "http://seankelly.biz/"},    

{"type": "home", "value": "http://seankelly.tv/"}    

]    

}

第二步   讀入Json資料

使用jsonFile讀入資料並註冊成表jsonPerson，然後定義一個查詢jsonQuery

scala>jsonFile("/home/hadoop/upload/class6/nestjson.json").registerTempTable("jsonPerson")

scala>val jsonQuery = sql("select * from jsonPerson")

第三步   檢視jsonQuery的schema

scala>jsonQuery.printSchema

第四步   檢視jsonQuery的整個執行計劃

scala>jsonQuery.queryExecution

1.4.5.2 讀取Parquet格式資料

Parquet資料放在配套資源/data/class6/wiki_parquet中，在以下測試中把檔案放到 /home/hadoop/upload/class6 路徑下

第一步   讀入Parquet資料

parquet檔案讀入並註冊成表parquetWiki，然後定義一個查詢parquetQuery

scala>parquetFile("/home/hadoop/upload/class6/wiki_parquet").registerTempTable("parquetWiki")

scala>val parquetQuery = sql("select * from parquetWiki")

有報錯但不影響使用

第二步   查詢parquetQuery的schema

scala>parquetQuery.printSchema

第三步   查詢parquetQuery的整個執行計劃

scala>parquetQuery.queryExecution

第四步   查詢取樣

scala>parquetQuery.takeSample(false,10,2)

1.4.5.3 讀取hive內建測試資料

在TestHive類中已經定義了大量的hive0.13的測試資料的表格式，如src、sales等等，在hive-console中可以直接使用；第一次使用的時候，hive-console會裝載一次。下面我們使用sales表看看其schema和整個執行計劃。

第一步   讀入測試資料並定義一個查詢hiveQuery

scala>val hiveQuery = sql("select * from sales")

第二步   檢視hiveQuery的schema

scala>hiveQuery.printSchema

第三步   檢視hiveQuery的整個執行計劃

scala>hiveQuery.queryExecution

第四步   其他SQL語句的執行計劃

scala>val hiveQuery = sql("select * from (select * from src limit 5) a limit 3")

scala>val hiveQuery = sql("select * FROM (select * FROM src) a")

scala>hiveQuery.where('key === 100).queryExecution.toRdd.collect

1.4.6 不同查詢的執行計劃

1.4.6.1 聚合查詢

scala>sql("select name, age,state as location from people").queryExecution

scala>sql("select name from (select name,state as location from people) a where location='CA'").queryExecution

scala>sql("select sum(age) from people").queryExecution

scala>sql("select sum(age) from people").toDebugString

scala>sql("select state,avg(age) from people group by state").queryExecution

scala>sql("select state,avg(age) from people group by state").toDebugString

1.4.6.2 Join操作

scala>sql("select a.name,b.name from people a join people b where a.name=b.name").queryExecution

scala>sql("select a.name,b.name from people a join people b where a.name=b.name").toDebugString

1.4.6.3 Distinct操作

scala>sql("select distinct a.name,b.name from people a join people b where a.name=b.name").queryExecution

scala>sql("select distinct a.name,b.name from people a join people b where a.name=b.name").toDebugString

1.4.7 優化

1.4.7.1 CombineFilters

CombineFilters就是合併Filter，在含有多個Filter時發生，如下查詢：

sql("select name from (select * from people where age >=19) a where a.age <30").queryExecution

上面的查詢，在Optimized的過程中，將age>=19和age<30這兩個Filter合併了，合併成((age>=19) && (age<30))。其實上面還做了一個其他的優化，就是project的下推，子查詢使用了表的所有列，而主查詢使用了列name，在查詢資料的時候子查詢優化成只查列name。

1.4.7.2 PushPredicateThroughProject

PushPredicateThroughProject就是project下推，和上面例子中的project一樣

sql("select name from (select name,state as location from people) a where location='CA'").queryExecution

1.4.7.3 ConstantFolding：

ConstantFolding是常量疊加，用於表示式。如下面的例子：

sql("select name,1+2 from people").queryExecution

2、SparkSQL調優

Spark是一個快速的記憶體計算框架，同時是一個並行運算的框架，在計算效能調優的時候，除了要考慮廣為人知的木桶原理外，還要考慮平行運算的Amdahl定理。

木桶原理又稱短板理論，其核心思想是：一隻木桶盛水的多少，並不取決於桶壁上最高的那塊木塊，而是取決於桶壁上最短的那塊。將這個理論應用到系統性能優化上，系統的最終效能取決於系統中效能表現最差的元件。例如，即使系統擁有充足的記憶體資源和CPU資源，但是如果磁碟I/O效能低下，那麼系統的總體效能是取決於當前最慢的磁碟I/O速度，而不是當前最優越的CPU或者記憶體。在這種情況下，如果需要進一步提升系統性能，優化記憶體或者CPU資源是毫無用處的。只有提高磁碟I/O效能才能對系統的整體效能進行優化。

Amdahl定理，一個電腦科學界的經驗法則，因吉恩·阿姆達爾而得名。它代表了處理器平行運算之後效率提升的能力。平行計算中的加速比是用並行前的執行速度和並行後的執行速度之比來表示的，它表示了在並行化之後的效率提升情況。阿姆達爾定律是固定負載（計算總量不變時）時的量化標準。可用公式：來表示。式中分別表示問題規模的序列分量（問題中不能並行化的那一部分）和並行分量，p表示處理器數量。當時，上式的極限是，其中。這意味著無論我們如何增大處理器數目，加速比是無法高於這個數的。

SparkSQL作為Spark的一個元件，在調優的時候，也要充分考慮到上面的兩個原理，既要考慮如何充分的利用硬體資源，又要考慮如何利用好分散式系統的平行計算。由於測試環境條件有限，本篇不能做出更詳盡的實驗資料來說明，只能在理論上加以說明。

2.1 並行性

SparkSQL在叢集中執行，將一個查詢任務分解成大量的Task分配給叢集中的各個節點來執行。通常情況下，Task的數量是大於叢集的並行度。比如前面第六章和第七章查詢資料時，shuffle的時候使用了預設的spark.sql.shuffle.partitions，即200個partition，也就是200個Task：

而實驗的叢集環境卻只能並行3個Task，也就是說同時只能有3個Task保持Running：

這時大家就應該明白了，要跑完這200個Task就要跑200/3=67批次。如何減少執行的批次呢？那就要儘量提高查詢任務的並行度。查詢任務的並行度由兩方面決定：叢集的處理能力和叢集的有效處理能力。

l對於Spark Standalone叢集來說，叢集的處理能力是由conf/spark-env中的SPARK_WORKER_INSTANCES引數、SPARK_WORKER_CORES引數決定的；而SPARK_WORKER_INSTANCES*SPARK_WORKER_CORES不能超過物理機器的實際CPU core；

l叢集的有效處理能力是指叢集中空閒的叢集資源，一般是指使用spark-submit或spark-shell時指定的--total-executor-cores，一般情況下，我們不需要指定，這時候，Spark Standalone叢集會將所有空閒的core分配給查詢，並且在Task輪詢執行過程中，Standalone叢集會將其他spark應用程式執行完後空閒出來的core也分配給正在執行中的查詢。

綜上所述，SparkSQL的查詢並行度主要和叢集的core數量相關，合理配置每個節點的core可以提高叢集的並行度，提高查詢的效率。

2.2 高效的資料格式

高效的資料格式，一方面是加快了資料的讀入速度，另一方面可以減少記憶體的消耗。高效的資料格式包括多個方面：

2.2.1 資料本地性

分散式計算系統的精粹在於移動計算而非移動資料，但是在實際的計算過程中，總存在著移動資料的情況，除非是在叢集的所有節點上都儲存資料的副本。移動資料，將資料從一個節點移動到另一個節點進行計算，不但消耗了網路IO，也消耗了磁碟IO，降低了整個計算的效率。為了提高資料的本地性，除了優化演算法（也就是修改spark記憶體，難度有點高），就是合理設定資料的副本。設定資料的副本，這需要通過配置引數並長期觀察執行狀態才能獲取的一個經驗值。

下面是Spark webUI監控Stage的一個圖：

lPROCESS_LOCAL是指讀取快取在本地節點的資料

lNODE_LOCAL是指讀取本地節點硬碟資料

lANY是指讀取非本地節點資料

l通常讀取資料PROCESS_LOCAL>NODE_LOCAL>ANY，儘量使資料以PROCESS_LOCAL或NODE_LOCAL方式讀取。其中PROCESS_LOCAL還和cache有關。

2.2.2 合適的資料型別

對於要查詢的資料，定義合適的資料型別也是非常有必要。對於一個tinyint可以使用的資料列，不需要為了方便定義成int型別，一個tinyint的資料佔用了1個byte，而int佔用了4個byte。也就是說，一旦將這資料進行快取的話，記憶體的消耗將增加數倍。在SparkSQL裡，定義合適的資料型別可以節省有限的記憶體資源。

2.2.3 合適的資料列

對於要查詢的資料，在寫SQL語句的時候，儘量寫出要查詢的列名，如Select a,b from tbl，而不是使用Select * from tbl；這樣不但可以減少磁碟IO，也減少快取時消耗的記憶體。

2.2.4 優的資料儲存格式

在查詢的時候，最終還是要讀取儲存在檔案系統中的檔案。採用更優的資料儲存格式，將有利於資料的讀取速度。檢視SparkSQL的Stage，可以發現，很多時候，資料讀取消耗佔有很大的比重。對於sqlContext來說，支援textFiile、SequenceFile、ParquetFile、jsonFile；對於hiveContext來說，支援AvroFile、ORCFile、Parquet File，以及各種壓縮。根據自己的業務需求，測試並選擇合適的資料儲存格式將有利於提高SparkSQL的查詢效率。

2.3 記憶體的使用

spark應用程式最糾結的地方就是記憶體的使用了，也是最能體現“細節是魔鬼”的地方。Spark的記憶體配置項有不少，其中比較重要的幾個是：

lSPARK_WORKER_MEMORY，在conf/spark-env.sh中配置SPARK_WORKER_MEMORY 和SPARK_WORKER_INSTANCES，可以充分的利用節點的記憶體資源，SPARK_WORKER_INSTANCES*SPARK_WORKER_MEMORY不要超過節點本身具備的記憶體容量；

lexecutor-memory，在spark-shell或spark-submit提交spark應用程式時申請使用的記憶體數量；不要超過節點的SPARK_WORKER_MEMORY；

lspark.storage.memoryFraction spark應用程式在所申請的記憶體資源中可用於cache的比例

lspark.shuffle.memoryFraction spark應用程式在所申請的記憶體資源中可用於shuffle的比例

在實際使用上，對於後兩個引數，可以根據常用查詢的記憶體消耗情況做適當的變更。另外，在SparkSQL使用上，有幾點建議：

l對於頻繁使用的表或查詢才進行快取，對於只使用一次的表不需要快取；

l對於join操作，優先快取較小的表；

l要多注意Stage的監控，多思考如何才能更多的Task使用PROCESS_LOCAL；

l要多注意Storage的監控，多思考如何才能Fraction cached的比例更多

2.4 合適的Task

對於SparkSQL，還有一個比較重要的引數，就是shuffle時候的Task數量，通過spark.sql.shuffle.partitions來調節。調節的基礎是spark叢集的處理能力和要處理的資料量，spark的預設值是200。Task過多，會產生很多的任務啟動開銷，Task多少，每個Task的處理時間過長，容易straggle。

2.5 其他的一些建議

優化的方面的內容很多，但大部分都是細節性的內容，下面就簡單地提提：

l  想要獲取更好的表示式查詢速度，可以將spark.sql.codegen設定為Ture；

l  對於大資料集的計算結果，不要使用collect() ,collect()就結果返回給driver，很容易撐爆driver的記憶體；一般直接輸出到分散式檔案系統中；

l  對於Worker傾斜，設定spark.speculation=true 將持續不給力的節點去掉；

l  對於資料傾斜，採用加入部分中間步驟，如聚合後cache，具體情況具體分析；

l  適當的使用序化方案以及壓縮方案；

l  善於利用叢集監控系統，將叢集的執行狀況維持在一個合理的、平穩的狀態；

l  善於解決重點矛盾，多觀察Stage中的Task，檢視最耗時的Task，查詢原因並改善；