大數據分析查詢引擎Impala

Impala是Cloudera公司主導開發的新型查詢系統，它提供SQL語義，能查詢存儲在Hadoop的HDFS和HBase中的PB級大數 據。已有的Hive系統雖然也提供了SQL語義，但由于Hive底層執行使用的是MapReduce引擎，仍然是一個批處理過程，難以滿足查詢的交互性。 相比之下，Impala的最大特點也是最大賣點就是它的快速。在介紹Impala之前需要先介紹Google的Dremel系統，因為Impala最開始 是參照 Dremel系統進行設計的。

Dremel是Google的交互式數據分析系統，它構建于Google的GFS（Google File System）等系統之上，支撐了Google的數據分析服務BigQuery等諸多服務。Dremel的技術亮點主要有兩個：一是實現了嵌套型數據的列 存儲；二是使用了多層查詢樹，使得任務可以在數千個節點上并行執行和聚合結果。列存儲在關系型數據庫中并不陌生，它可以減少查詢時處理的數據量，有效提升 查詢效率。Dremel的列存儲的不同之處在于它針對的并不是傳統的關系數據，而是嵌套結構的數據。Dremel可以將一條條的嵌套結構的記錄轉換成列存 儲形式，查詢時根據查詢條件讀取需要的列，然后進行條件過濾，輸出時再將列組裝成嵌套結構的記錄輸出，記錄的正向和反向轉換都通過高效的狀態機實現。另 外，Dremel的多層查詢樹則借鑒了分布式搜索引擎的設計，查詢樹的根節點負責接收查詢，并將查詢分發到下一層節點，底層節點負責具體的數據讀取和查詢 執行，然后將結果返回上層節點。

Impala是Cloudera在受到Google的Dremel啟發下開發的實時交互SQL大數據查詢工具，Impala沒有再使用緩慢的 Hive+MapReduce批處理，而是通過使用與商用并行關系數據庫中類似的分布式查詢引擎（由Query Planner、Query Coordinator和Query Exec Engine三部分組成），可以直接從HDFS或HBase中用SELECT、JOIN和統計函數查詢數據，從而大大降低了延遲。Impala其實就是 Hadoop的Dremel，Impala使用的列存儲格式是Parquet。Parquet實現了Dremel中的列存儲，未來還將支持 Hive并添加字典編碼、游程編碼等功能。Impala的系統架構如圖所示。Impala使用了Hive的SQL接口（包括SELECT、 INSERT、Join等操作），但目前只實現了Hive的SQL語義的子集（例如尚未對UDF提供支持），表的元數據信息存儲在Hive的 Metastore中。StateStore是Impala的一個子服務，用來監控集群中各個節點的健康狀況，提供節點注冊、錯誤檢測等功能。 Impala在每個節點運行了一個后臺服務Impalad，Impalad用來響應外部請求，并完成實際的查詢處理。Impalad主要包含Query Planner、Query Coordinator和Query Exec Engine三個模塊。QueryPalnner接收來自SQL APP和ODBC的查詢，然后將查詢轉換為許多子查詢，Query Coordinator將這些子查詢分發到各個節點上，由各個節點上的Query Exec Engine負責子查詢的執行，最后返回子查詢的結果，這些中間結果經過聚集之后最終返回給用戶。

Impala主要由Impalad， State Store和CLI組成。

Impalad

與DataNode運行在同一節點上，由Impalad進程表示，它接收客戶端的查詢請求（接收查詢請求的Impalad為 Coordinator，Coordinator通過JNI調用java前端解釋SQL查詢語句，生成查詢計劃樹，再通過調度器把執行計劃分發給具有相應 數據的其它Impalad進行執行），讀寫數據，并行執行查詢，并把結果通過網絡流式的傳送回給Coordinator，由Coordinator返回給 客戶端。同時Impalad也與State Store保持連接，用于確定哪個Impalad是健康和可以接受新的工作。在Impalad中啟動三個ThriftServer: beeswax_server（連接客戶端），hs2_server（借用Hive元數據）， be_server（Impalad內部使用）和一個ImpalaServer服務。每個impalad實例會接收、規劃并調節來自ODBC或 Impala Shell等客戶端的查詢。每個impalad實例會充當一個Worker，處理由其它impalad實例分發出來的查詢片段(query fragments)。客戶端可以隨便連接到任意一個impalad實例，被連接的impalad實例將充當本次查詢的協調者（Ordinator)，將 查詢分發給集群內的其它impalad實例進行并行計算。當所有計算完畢時，其它各個impalad實例將會把各自的計算結果發送給充當 Ordinator的impalad實例，由這個Ordinator實例把結果返回給客戶端。每個impalad進程可以處理多個并發請求。

Impala State Store

跟蹤集群中的Impalad的健康狀態及位置信息，由statestored進程表示，它通過創建多個線程來處理Impalad的注冊訂閱和與各 Impalad保持心跳連接，各Impalad都會緩存一份State Store中的信息，當State Store離線后（Impalad發現State Store處于離線時，會進入recovery模式，反復注冊，當State Store重新加入集群后，自動恢復正常，更新緩存數據）因為Impalad有State Store的緩存仍然可以工作，但會因為有些Impalad失效了，而已緩存數據無法更新，導致把執行計劃分配給了失效的Impalad，導致查詢失敗。

• 用于協調各個運行impalad的實例之間的信息關系，Impala正是通過這些信息去定位查詢請求所要的數據。換句話說，state store的作用主要為跟蹤各個impalad實例的位置和狀態，讓各個impalad實例以集群的方式運行起來。
• 與 HDFS的NameNode不一樣，雖然State Store一般只安裝一份，但一旦State Store掛掉了，各個impalad實例卻仍然會保持集群的方式處理查詢請求，只是無法將各自的狀態更新到State Store中，如果這個時候新加入一個impalad實例，則新加入的impalad實例不為現有集群中的其他impalad實例所識別（事實上，經筆者 測試，如果impalad啟動在statestored之后，根本無法正常啟動，因為impalad啟動時是需要指定statestored的主機信息 的）。然而，State Store一旦重啟，則所有State Store所服務的各個impalad實例（包括state store掛掉期間新加入的impalad實例）的信息（由impalad實例發給state store）都會進行重建。

CLI （Impala shell）

提供給用戶查詢使用的命令行工具（Impala Shell使用python實現），同時Impala還提供了Hue，JDBC， ODBC使用接口。該客戶端工具提供一個交互接口，供使用者發起數據查詢或管理任務，比如連接到impalad。這些查詢請求會傳給ODBC這個標準查詢 接口。說白了，就是一個命令行客戶端。

與Hive的關系

Impala與Hive都是構建在Hadoop之上的數據查詢工具各有不同的側重適應面，但從客戶端使用來看Impala與Hive有很多的共同 之處，如數據表元數據、ODBC/JDBC驅動、SQL語法、靈活的文件格式、存儲資源池等。Impala與Hive在Hadoop中的關系下圖所示。 Hive適合于長時間的批處理查詢分析，而Impala適合于實時交互式SQL查詢，Impala給數據分析人員提供了快速實驗、驗證想法的大數據分析工 具。可以先使用hive進行數據轉換處理，之后使用Impala在Hive處理后的結果數據集上進行快速的數據分析。

SQL 支持度：

支持SQL92中的大部分select語句， 以及SQL2003標準中的分析函數。 不支持DELETE和UPDATE， 但是支持批量裝載數據（insert into select, LOAD DATA) 和批量刪除數據（drop partition）。除此之外， 用戶也可直接操作HDFS文件實現數據裝載和清理。

查詢執行

impalad分為frontend和backend兩個層次， frondend用java實現（通過JNI嵌入impalad）， 負責查詢計劃生成， 而backend用C++實現， 負責查詢執行。

frontend生成查詢計劃分為兩個階段：（1）生成單機查詢計劃，單機執行計劃與關系數據庫執行計劃相同，所用查詢優化方法也類似。（2）生成分布式查詢計劃。 根據單機執行計劃， 生成真正可執行的分布式執行計劃，降低數據移動， 盡量把數據和計算放在一起。

上圖是SQL查詢例子， 該SQL的目標是在三表join的基礎上算聚集， 并按照聚集列排序取topN。 impala的查詢優化器支持代價模型： 利用表和分區的cardinality，每列的distinct值個數等統計數據， impala可估算執行計劃代價， 并生成較優的執行計劃。 上圖左邊是frontend查詢優化器生成的單機查詢計劃， 與傳統關系數據庫不同， 單機查詢計劃不能直接執行， 必須轉換成如圖右半部分所示的分布式查詢計劃。 該分布式查詢計劃共分成6個segment（圖中彩色無邊框圓角矩形）， 每個segment是可以被單臺服務器獨立執行的計劃子樹。

impala支持兩種分布式join方式， 表廣播和哈希重分布：表廣播方式保持一個表的數據不動， 將另一個表廣播到所有相關節點（圖中t3）； 哈希重分布的原理是根據join字段哈希值重新分布兩張表數據(譬如圖中t1和t2）。分布式計劃中的聚集函數分拆為兩個階段執行。第一步針對本地數據進 行分組聚合（Pre-AGG）以降低數據量， 并進行數據重分步， 第二步， 進一步匯總之前的聚集結果（mergeAgg）計算出最終結果。 與聚集函數類似， topN也是分為兩個階段執行， （1）本地排序取topN，以降低數據量； （2） merge sort得到最終topN結果。

Backend從frontend接收plan segment并執行， 執行性能非常關鍵，impala采取的查詢性能優化措施有

• 向量執行。 一次getNext處理一批記錄， 多個操作符可以做pipeline。
• LLVM編譯執行， CPU密集型查詢效率提升5倍以上。
• IO本地化。 利用HDFS short-circuit local read功能，實現本地文件讀取
• Parquet列存，相比其他格式性能最高提升5倍。

資源管理

impala通常與MR等離線任務運行在一個集群上， 通過YARN統一管理資源， 如何同時滿足交互式查詢和離線查詢兩種需求具有較大挑戰性。 YARN通過全局唯一的Resource Mananger調度資源， 好處是RM擁有整個集群全局信息，能做出更好調度決策， 缺點是資源分配的性能不足。 Impala每個查詢都需要分配資源， 當每秒查詢數上千時， YARN資源分配的響應時間變的很長， 影響到查詢性能。 目前通過兩個措施解決這個問題：（1）引入快速、非集中式的查詢準入機制， 控制查詢并發度。（2）LLAM（low latency application master）通過緩存資源， 批量分配，增量分配等方式實現降低資源分配延時

Impala 相對于Hive所使用的優化技術

• 沒有使用MapReduce進行并行計算，雖然MapReduce是非常好的并行計算框架，但它更多的面向批處理模式，而不是面向交互式的 SQL執行。與MapReduce相比：Impala把整個查詢分成一執行計劃樹，而不是一連串的MapReduce任務，在分發執行計劃 后，Impala使用拉式獲取數據的方式獲取結果，把結果數據組成按執行樹流式傳遞匯集，減少的了把中間結果寫入磁盤的步驟，再從磁盤讀取數據的開銷。 Impala使用服務的方式避免每次執行查詢都需要啟動的開銷，即相比Hive沒了MapReduce啟動時間。
• 使用LLVM產生運行代碼，針對特定查詢生成特定代碼，同時使用Inline的方式減少函數調用的開銷，加快執行效率。
• 充分利用可用的硬件指令（2）。
• 更好的IO調度，Impala知道數據塊所在的磁盤位置能夠更好的利用多磁盤的優勢，同時Impala支持直接數據塊讀取和本地代碼計算checksum。
• 通過選擇合適的數據存儲格式可以得到最好的性能（Impala支持多種存儲格式）。
• 最大使用內存，中間結果不寫磁盤，及時通過網絡以stream的方式傳遞。

Impala 與Hive的異同

相同點：

• 數據存儲：使用相同的存儲數據池都支持把數據存儲于HDFS, HBase。
• 元數據：兩者使用相同的元數據。
• SQL解釋處理：比較相似都是通過詞法分析生成執行計劃。

不同點：

執行計劃：

• Hive: 依賴于MapReduce執行框架，執行計劃分成 map->shuffle->reduce->map->shuffle->reduce…的模型。如果一個Query會 被編譯成多輪MapReduce，則會有更多的寫中間結果。由于MapReduce執行框架本身的特點，過多的中間過程會增加整個Query的執行時間。
• Impala: 把執行計劃表現為一棵完整的執行計劃樹，可以更自然地分發執行計劃到各個Impalad執行查詢，而不用像Hive那樣把它組合成管道型的 map->reduce模式，以此保證Impala有更好的并發性和避免不必要的中間sort與shuffle。

數據流：

• Hive: 采用推的方式，每一個計算節點計算完成后將數據主動推給后續節點。
• Impala: 采用拉的方式，后續節點通過getNext主動向前面節點要數據，以此方式數據可以流式的返回給客戶端，且只要有1條數據被處理完，就可以立即展現出來，而不用等到全部處理完成，更符合SQL交互式查詢使用。

內存使用：

• Hive: 在執行過程中如果內存放不下所有數據，則會使用外存，以保證Query能順序執行完。每一輪MapReduce結束，中間結果也會寫入HDFS中，同樣由于MapReduce執行架構的特性，shuffle過程也會有寫本地磁盤的操作。
• Impala: 在遇到內存放不下數據時，當前版本0.1是直接返回錯誤，而不會利用外存，以后版本應該會進行改進。這使用得Impala目前處理Query會受到一定的 限制，最好還是與Hive配合使用。Impala在多個階段之間利用網絡傳輸數據，在執行過程不會有寫磁盤的操作（insert除外）。

調度：

• Hive: 任務調度依賴于Hadoop的調度策略。
• Impala: 調度由自己完成，目前只有一種調度器simple-schedule，它會盡量滿足數據的局部性，掃描數據的進程盡量靠近數據本身所在的物理機器。調度器 目前還比較簡單，在SimpleScheduler::GetBackend中可以看到，現在還沒有考慮負載，網絡IO狀況等因素進行調度。但目前 Impala已經有對執行過程的性能統計分析，應該以后版本會利用這些統計信息進行調度吧。

容錯：

• Hive: 依賴于Hadoop的容錯能力。
• Impala: 在查詢過程中，沒有容錯邏輯，如果在執行過程中發生故障，則直接返回錯誤（這與Impala的設計有關，因為Impala定位于實時查詢，一次查詢失敗， 再查一次就好了，再查一次的成本很低）。但從整體來看，Impala是能很好的容錯，所有的Impalad是對等的結構，用戶可以向任何一個 Impalad提交查詢，如果一個Impalad失效，其上正在運行的所有Query都將失敗，但用戶可以重新提交查詢由其它Impalad代替執行，不 會影響服務。對于State Store目前只有一個，但當State Store失效，也不會影響服務，每個Impalad都緩存了State Store的信息，只是不能再更新集群狀態，有可能會把執行任務分配給已經失效的Impalad執行，導致本次Query失敗。

適用面：

• Hive: 復雜的批處理查詢任務，數據轉換任務。
• Impala：實時數據分析，因為不支持UDF，能處理的問題域有一定的限制，與Hive配合使用,對Hive的結果數據集進行實時分析。

Impala 的優缺點

優點：

• 支持SQL查詢，快速查詢大數據。
• 可以對已有數據進行查詢，減少數據的加載，轉換。
• 多種存儲格式可以選擇（Parquet, Text, Avro, RCFile, SequeenceFile）。
• 可以與Hive配合使用。

缺點：

• 不支持用戶定義函數UDF。
• 不支持text域的全文搜索。
• 不支持Transforms。
• 不支持查詢期的容錯。
• 對內存要求高。

在Cloudera的測試中，Impala的查詢效率比Hive有數量級的提升。從技術角度上來看，Impala之所以能有好的性能，主要有以下幾方面的原因。

• Impala不需要把中間結果寫入磁盤，省掉了大量的I/O開銷。
• 省掉了MapReduce作業啟動的開銷。MapReduce啟動task的速度很慢（默認每個心跳間隔是3秒鐘），Impala直接通過相應的服務進程來進行作業調度，速度快了很多。
• Impala完全拋棄了MapReduce這個不太適合做SQL查詢的范式，而是像Dremel一樣借鑒了MPP并行數據庫的思想另起爐灶，因此可做更多的查詢優化，從而省掉不必要的shuffle、sort等開銷。
• 通過使用LLVM來統一編譯運行時代碼，避免了為支持通用編譯而帶來的不必要開銷。
• 用C++實現，做了很多有針對性的硬件優化，例如使用SSE指令。
• 使用了支持Data locality的I/O調度機制，盡可能地將數據和計算分配在同一臺機器上進行，減少了網絡開銷。

雖然Impala是參照Dremel來實現的，但它也有一些自己的特色，例如Impala不僅支持Parquet格式，同時也可以直接處理文本、 SequenceFile等Hadoop中常用的文件格式。另外一個更關鍵的地方在于，Impala是開源的，再加上Cloudera在Hadoop領域 的領導地位，其生態圈有很大可能會在將來快速成長。

可以預見，在不久的未來，Impala很可能像之前的Hadoop和Hive一樣在大數 據處理領域大展拳腳。Cloudera自己也說期待未來Impala能完全取代Hive。當然，用戶從Hive上遷移到Impala上來是需要時間的。需 要說明的是，Impala并不是用來取代已有的MapReduce系統，而是作為MapReduce的一個強力補充。總的來說，Impala適合用來處理 輸出數據適中或比較小的查詢，而對于大數據量的批處理任務，MapReduce依然是更好的選擇。另外一個消息是，Cloudera里負責Impala的 架構師Marcel Komacker就曾在Google負責過F1系統的查詢引擎開發，可見Google確實為大數據的流行出錢出力。

Impala 與Shark，Drill等的比較

開源組織Apache也發起了名為Drill的項目來實現Hadoop上的Dremel，目前該項目正在開發當中，相關的文檔和代碼還不多，可以 說暫時還未對Impala構成足夠的威脅。從Quora上的問答來看，Cloudera有7-8名工程師全職在Impala項目上，而相比之下Drill 目前的動作稍顯遲鈍。具體來說，截止到2012年10月底，Drill的代碼庫里實現了query parser, plan parser，及能對JSON格式的數據進行掃描的plan evaluator；而Impala同期已經有了一個比較完畢的分布式query execution引擎，并對HDFS和HBase上的數據讀入，錯誤檢測，INSERT的數據修改，LLVM動態翻譯等都提供了支持。當然，Drill 作為Apache的項目，從一開始就避免了某個vendor的一家獨大，而且對所有Hadoop流行的發行版都會做相應的支持，不像Impala只支持 Cloudera自己的發行版CDH。從長遠來看，誰會占據上風還真不一定。

除此之外，加州伯克利大學AMPLab也開發了名為Shark的大數據分析系統。從長遠目標來看，Shark想成為一個既支持大數據SQL查詢， 又能支持高級數據分析任務的一體化數據處理系統。從技術實現的角度上來看，Shark基于Scala語言的算子推導實現了良好的容錯機制，因此對失敗了的 長任務和短任務都能從上一個“快照點”進行快速恢復。相比之下，Impala由于缺失足夠強大的容錯機制，其上運行的任務一旦失敗就必須“從頭來過”，這 樣的設計必然會在性能上有所缺失。而且Shark是把內存當作第一類的存儲介質來做的系統設計，所以在處理速度上也會有一些優勢。實際上，AMPLab最 近對Hive，Impala，Shark及Amazon采用的商業MPP數據庫Redshift進行了一次對比試驗，在Scan Query，Aggregation Query和Join Query三種類型的任務中對它們進行了比較。圖2就是AMPLab報告中Aggregation Query的性能對比。在圖中我們可以看到，商業版本的Redshift的性能是最好的， Impala和Shark則各有勝負，且兩者都比Hive的性能高出了一大截。

其實對大數據分析的項目來說，技術往往不是最關鍵的。例如Hadoop中的MapReduce和HDFS都是源于Google，原創性較少。事實 上，開源項目的生態圈，社區，發展速度等，往往在很大程度上會影響Impala和Shark等開源大數據分析系統的發展。就像Cloudera一開始就決 定會把Impala開源，以期望利用開源社區的力量來推廣這個產品；Shark也是一開始就開源了出來，更不用說Apache的Drill更是如此。說到 底還是誰的生態系統更強的問題。技術上一時的領先并不足以保證項目的最終成功。雖然最后那一款產品會成為事實上的標準還很難說，但是，我們唯一可以確定并 堅信的一點是，大數據分析將隨著新技術的不斷推陳出新而不斷普及開來，這對用戶永遠都是一件幸事。舉個例子，如果讀者注意過下一代 Hadoop（YARN）的發展的話就會發現，其實YARN已經支持MapReduce之外的計算范式（例如Shark，Impala等），因此將來 Hadoop將可能作為一個兼容并包的大平臺存在，在其上提供各種各樣的數據處理技術，有應對秒量級查詢的，有應對大數據批處理的，各種功能應有盡有，滿 足用戶各方面的需求。

未來展望

其實除了Impala，Shark，Drill這樣的開源方案外，像Oracle，EMC等傳統廠商也沒在坐以待斃等著自己的市場被開源軟件侵 吞。像EMC就推出了HAWQ系統，并號稱其性能比之Impala快上十幾倍，而前面提到的Amazon的Redshift也提供了比Impala更好的 性能。雖然說開源軟件因為其強大的成本優勢而擁有極其強大的力量，但是傳統數據庫廠商仍會嘗試推出性能、穩定性、維護服務等指標上更加強大的產品與之進行 差異化競爭，并同時參與開源社區、借力開源軟件來豐富自己的產品線、提升自己的競爭力，并通過更多的高附加值服務來滿足某些消費者需求。畢竟，這些廠商往 往已在并行數據庫等傳統領域積累了大量的技術和經驗，這些底蘊還是非常深厚的。甚至現在還有像NuoDB（一個創業公司）這樣號稱即支持ACID，又有 Scalability的NewSQL系統出來。總的來看，未來的大數據分析技術將會變得越來越成熟、越來越便宜、越來越易用；相應的，用戶將會更容易更 方便地從自己的大數據中挖掘出有價值的商業信息。

參考資料

• http://impala.io/