SparkSQL－從DataFrame說起 – 有態(tài)度的HBase/Spark/BigData

無名小卒917 2018-01-24

展開全文

本著更好地理解大數(shù)據(jù)生態(tài)圈的本意以及工作的需要，前段時間熟悉了SQL查詢引擎SparkSQL、Hadoop文件格式Parquet/CarbonData、大數(shù)據(jù)基準測試標準TPCDS/TPCH等相關知識，后續(xù)將會陸續(xù)整理出相關的內(nèi)容；所有分享內(nèi)容都是參考相關資料完成，文中很多細節(jié)都是在閱讀相關資料時的所感所悟，只希望能夠及時記錄下來，以免遺忘！另外，不可避免會有一些紕漏，還忘客官能夠批判性閱讀，討論交流！當然，HBase相關博客還會繼續(xù)更新；

SparkSQL 歷史回顧

對SparkSQL了解的童鞋或多或少聽說過Shark，不錯，Shark就是SparkSQL的前身。2011的時候，Hive可以說是SQL On Hadoop的唯一選擇，負責將SQL解析成MR任務運行在大數(shù)據(jù)上，實現(xiàn)交互式查詢、報表等功能。就在那個時候，Spark社區(qū)的小伙伴就意識到可以使用Spark作為執(zhí)行引擎替換Hive中的MR，這樣可以使Hive的執(zhí)行效率得到極大提升。這個思想的產(chǎn)物就是Shark，所以從實現(xiàn)功能上來看，Shark更像一個Hive On Spark實現(xiàn)版本。

改造完成剛開始，Shark確實比Hive的執(zhí)行效率有了極大提升。然而，隨著改造的深入，發(fā)現(xiàn)因為Shark繼承了大量Hive代碼導致添加優(yōu)化規(guī)則等變得異常困難，優(yōu)化的前景不再那么樂觀。在意識到這個問題之后，Spark社區(qū)經(jīng)過一段時間激烈的思想斗爭之后，還是毅然決然的在2014年徹底放棄了Shark，轉向SparkSQL。

因此可以理解為SparkSQL是一個全新的項目，接下來將會帶大家一起走近SparkSQL的世界，從SparkSQL體系的最頂端走向最底層，尋根問底，深入理解SparkSQL是如何工作的。

SparkSQL 體系結構

SparkSQL體系結構如下圖所示，整體由上到下分為三層：編程模型層、執(zhí)行任務優(yōu)化層以及任務執(zhí)行引擎層，其中SparkSQL編程模型可以分為SQL和DataFrame兩種；執(zhí)行計劃優(yōu)化又稱為Catalyst，該模塊負責將SQL語句解析成AST（邏輯執(zhí)行計劃），并對原始邏輯執(zhí)行計劃進行優(yōu)化，優(yōu)化規(guī)則分為基于規(guī)則的優(yōu)化策略和基于代價的優(yōu)化策略兩種，最終輸出優(yōu)化后的物理執(zhí)行計劃；任務執(zhí)行引擎就是Spark內(nèi)核，負責根據(jù)物理執(zhí)行計劃生成DAG，在任務調(diào)度系統(tǒng)的管理下分解為任務集并分發(fā)到集群節(jié)點上加載數(shù)據(jù)運行，Tungsten基于對內(nèi)存和CPU的性能優(yōu)化，使得Spark能夠更好地利用當前硬件條件提升性能，詳情可以閱讀 Project Tungsten: Bringing Apache Spark Closer to Bare Metal。

SparkSQL系列文章會按照體系結構由上至下詳細地進行說明，本篇下面會重點講解編程接口DataFrame，后面會利用M篇文章分析Catalyst的工作原理，再后面會利用N篇文章分析Spark內(nèi)核工作原理。

SparkSQL 編程模型－ DataFrame

說到計算模型，批處理計算從最初提出一直到現(xiàn)在，一共經(jīng)歷了兩次大的變革，第一次變革是從MR編程模式到RDD編程模型，第二次則是從RDD編程模式進化到DataFrame模式。

第一次變革：MR編程模型 -> DAG編程模型

和MR計算模型相比，DAG計算模型有很多改進：

1. 可以支持更多的算子，比如filter算子、sum算子等，不再像MR只支持map和reduce兩種

2. 更加靈活的存儲機制，RDD可以支持本地硬盤存儲、緩存存儲以及混合存儲三種模式，用戶可以進行選擇。而MR目前只支持HDFS存儲一種模式。很顯然，HDFS存儲需要將中間數(shù)據(jù)存儲三份，而RDD則不需要，這是DAG編程模型效率高的一個重要原因之一。

3. DAG模型帶來了更細粒度的任務并發(fā)，不再像MR那樣每次起個任務就要起個JVM進程，重死了；另外，DAG模型帶來了另一個利好是很好的容錯性，一個任務即使中間斷掉了，也不需要從頭再來一次。

4. 延遲計算機制一方面可以使得同一個stage內(nèi)的操作可以合并到一起落在一塊數(shù)據(jù)上，而不再是所有數(shù)據(jù)先執(zhí)行a操作、再掃描一遍執(zhí)行b操作，太浪費時間。另一方面給執(zhí)行路徑優(yōu)化留下了可能性，隨便你怎么優(yōu)化…

所有這些改進使得DAG編程模型相比MR編程模型，性能可以有10～100倍的提升！然而，DAG計算模型就很完美嗎？要知道，用戶手寫的RDD程序基本或多或少都會有些問題，性能也肯定不會是最優(yōu)的。如果沒有一個高手指點或者優(yōu)化，性能依然有很大的優(yōu)化潛力。這就是促成了第二次變革，從DAG編程模型進化到DataFrame編程模型。

第二次變革：DAG編程模型 -> DataFrame編程模型

相比RDD，DataFrame增加了scheme概念，從這個角度看，DataFrame有點類似于關系型數(shù)據(jù)庫中表的概念?？梢愿鶕?jù)下圖對比RDD與DataFrame數(shù)據(jù)結構的差別：

直觀上看，DataFrame相比RDD多了一個表頭，這個小小的變化帶來了很多優(yōu)化的空間：

1. RDD中每一行紀錄都是一個整體，因此你不知道內(nèi)部數(shù)據(jù)組織形式，這就使得你對數(shù)據(jù)項的操作能力很弱。表現(xiàn)出來就是支持很少的而且是比較粗粒度的算子，比如map、filter算子等。而DataFrame將一行切分了多個列，每個列都有一定的數(shù)據(jù)格式，這與數(shù)據(jù)庫表模式就很相似了，數(shù)據(jù)粒度相比更細，因此就能支持更多更細粒度的算子，比如select算子、groupby算子、where算子等。更重要的，后者的表達能力要遠遠強于前者，比如同一個功能用RDD和DataFrame實現(xiàn)：

2. DataFrame的Schema的存在，數(shù)據(jù)項的轉換也都將是類型安全的，這對于較為復雜的數(shù)據(jù)計算程序的調(diào)試是十分有利的，很多數(shù)據(jù)類型不匹配的問題都可以在編譯階段就被檢查出來，而對于不合法的數(shù)據(jù)文件，DataFrame也具備一定分辨能力。

3. DataFrame schema的存在，開辟了另一種數(shù)據(jù)存儲形式：列式數(shù)據(jù)存儲。列式存儲是相對于傳統(tǒng)的行式存儲而言的，簡單來講，就是將同一列的所有數(shù)據(jù)物理上存儲在一起。對于列式存儲和行式存儲可以參考下圖：

列式存儲有兩個重要的作用，首先，同一種類型的數(shù)據(jù)存儲在一起可以很好的提升數(shù)據(jù)壓縮效率，因為越“相似”的數(shù)據(jù)，越容易壓縮。數(shù)據(jù)壓縮可以減少存儲空間需求，還可以減少數(shù)據(jù)傳輸過程中的帶寬需求，這對于類似于Spark之類的大內(nèi)存計算引擎來講，會帶來極大的益處；另外，列式存儲還可以有效減少查詢過程中的實際IO，大數(shù)據(jù)領域很多OLAP查詢業(yè)務通常只會檢索部分列值，而不是粗暴的select * ，這樣列式存儲可以有效執(zhí)行’列值裁剪’，將不需要查找的列直接跳過。

4. DAG編程模式都是用戶自己寫RDD scala程序，自己寫嘛，必然或多或少會有性能提升的空間！而DataFrame編程模式集成了一個優(yōu)化神奇－Catalyst，這玩意類似于MySQL的SQL優(yōu)化器，負責將用戶寫的DataFrame程序進行優(yōu)化得到最優(yōu)的執(zhí)行計劃（下文會講），比如最常見的謂詞下推優(yōu)化。很顯然，優(yōu)化后的執(zhí)行計劃相比于手寫的執(zhí)行計劃性能當然會來的好一些。下圖是官方給出來的測試對比數(shù)據(jù)（測試過程是在10billion數(shù)據(jù)規(guī)模下進行過濾聚合）：

個人覺得，RDD和DataFrame的關系類似于匯編語言和Java語言的關系，同一個功能，如果你用匯編實現(xiàn)的話，一方面會寫的很長，另一方面寫的代碼可能還不是最優(yōu)的，可謂是又臭又長。而Java語言有很多高級語意，可以很方便的實現(xiàn)相關功能，另一方面經(jīng)過JVM優(yōu)化后會更加高效。