大數據時代快速SQL引擎-Impala

時間 2019-11-17

標籤數據時代快速 sql 引擎 impala 欄目 SQL 简体版

原文原文鏈接

背景

隨着大數據時代的到來，Hadoop在過去幾年以接近統治性的方式包攬的ETL和數據分析查詢的工做，你們也無心間的想往大數據方向靠攏，即便天天數據也就幾10、幾百M也要放到Hadoop上做分析，只會拔苗助長，可是當面對真正的Big Data的時候，Hadoop就會暴露出它對於數據分析查詢支持的弱點。甚至出現《MapReduce: 一個巨大的倒退》此類極端的吐槽，這也怪不得Hadoop，畢竟它的設計就是爲了批處理，使用用MR的編程模型來實現SQL查詢，性能確定不如意。因此一般我也只是把Hive當作可以提供將SQL語義轉換成MR任務的工具，尤爲在作ETL的時候。html

在Dremel論文發表以後，開源社區涌現出了一批基於MPP架構的SQL-on-Hadoop(HDFS)查詢引擎，典型表明有Apache Impala、Presto、Apache Drill、Apache HAWQ等，看上去這些查詢引擎提供的功能和實現方式也都大同小異，本文將基於Impala的使用和實現介紹日益發展的基於HDFS的MPP數據查詢引擎。前端

Impala介紹

Apache Impala是由Cloudera開發並開源的一款基於HDFS/Hbase的MPP SQL引擎，它擁有和Hadoop同樣的可擴展性、它提供了類SQL（類Hsql）語法，在多用戶場景下也能擁有較高的響應速度和吞吐量。它是由Java和C++實現的，Java提供的查詢交互的接口和實現，C++實現了查詢引擎部分，除此以外，Impala還可以共享Hive Metastore（這逐漸變成一種標準），甚至能夠直接使用Hive的JDBC jar和beeline等直接對Impala進行查詢、支持豐富的數據存儲格式（Parquet、Avro等），固然除了有比較明確的理由，Parquet老是使用Impala的第一選擇。web

從用戶視角

能夠將Impala這類系統的用戶分爲兩類，一類是負責數據導入和管理的數據開發同窗，另外一類則是執行查詢的數據分析師同窗，前者一般須要將數據存儲到HDFS，經過CREATE TABLE的方式建立與數據match的schema，而後經過load data或者add partition的方式將表和數據關聯起來，這一些流程串起來仍是挺麻煩的，可是多虧了Hive，因爲Impala能夠共享Hive的MetaStore，這樣就可使用Hive完成此類ETL工做，而後將數據查詢的工做交給Impala，大大簡化工做流程（據我所知畢竟大部分數據開發同窗仍是比較熟悉Hive）。接下來對於數據分析師而言就是如何編寫正確的SQ以表達他們的查詢、分析需求，這也是它們最拿手的了，Impala一般能夠在TB級別的數據上提供秒級的查詢速度，因此使用起來可能讓你從Hive的龜速響應一下提高到指望的速度。sql

Impala除了支持簡單類型以外，還支持String、timestamp、decimal等多種類型，用戶還能夠對於特殊的邏輯實現自定義函數（UDF）和自定義聚合函數（UDAF），前者可使用Java和C++實現，後者目前僅支持C++實現，除此以外的schema操做均可以在Hive上實現，因爲Impala的存儲由HDFS實現，所以不可以實現update、delete語句，若是有此類需求，仍是須要從新計算整個分區的數據而且覆蓋老數據，這點對於修改的實時性要求比較高的需求仍是不能知足的，若是有此類需求仍是期待Kudu的支持吧，或者嘗試一下傳統的MPP數據庫，例如GreenPlum。數據庫

當完成數據導入以後，用戶須要執行COMPUTE STATSapache

系統架構

從用戶的使用方式上來看，Impala和Hive仍是很類似的，而且能夠共享一份元數據，這也大大簡化了接入流程，下面咱們從實現的角度來看一下Impala是如何工做的。下圖展現了Impala的系統架構和查詢的執行流程。編程

從上圖能夠看出，Impala自身包含三個模塊：Impalad、Statestore和Catalog，除此以外它還依賴Hive Metastore和HDFS，其中Imapalad負責接受用戶的查詢請求，也意味着用戶的能夠將請求發送給任意一個Impalad進程，該進程在本次查詢充當協調者（coordinator）的做用，生成執行計劃而且分發到其它的Impalad進程執行，最終聚集結果返回給用戶，而且對於當前Impalad和其它Impalad進程而言，他們同時也是本次查詢的執行者，完成數據讀取、物理算子的執行並將結果返回給協調者Impalad。這種無中心查詢節點的設計可以最大程度的保證容錯性而且很容易作負載均衡。正如圖中展現的同樣，一般每個HDFS的DataNode上部署一個Impalad進程，因爲HDFS存儲數據一般是多副本的，因此這樣的部署能夠保證數據的本地性，查詢儘量的從本地磁盤讀取數據而非網絡，從這點能夠推斷出Impalad對於本地數據的讀取應該是經過直接讀本地文件的方式，而非調用HDFS的接口。爲了實現查詢分割的子任務能夠作到儘量的本地數據讀取，Impalad須要從Metastore中獲取表的數據存儲路徑，而且從NameNode中獲取每個文件的數據塊分佈。後端

Catalog服務提供了元數據的服務，它以單點的形式存在，它既能夠從外部系統（例如HDFS NameNode和Hive Metastore）拉取元數據，也負責在Impala中執行的DDL語句提交到Metatstore，因爲Impala沒有update/delete操做，因此它不須要對HDFS作任何修改。以前咱們介紹過有兩種方式向Impala中導入數據（DDL）——經過hive或者impala，若是經過hive則改變的是Hive metastore的狀態，此時須要經過在Impala中執行REFRESH以通知元數據的更新，而若是在impala中操做則Impalad會將該更新操做通知Catalog，後者經過廣播的方式通知其它的Impalad進程。默認狀況下Catalog是異步加載元數據的，所以查詢可能須要等待元數據加載完成以後才能進行（第一次加載）。該服務的存在將元數據從Impalad進程中獨立出來，能夠簡化Impalad的實現，下降Impalad之間的耦合。緩存

除了Catalog服務，Impala還提供了StateStore服務完成兩個工做：消息訂閱服務和狀態監測功能。Catalog中的元數據就是經過StateStore服務進行廣播分發的，它實現了一個Pub-Sub服務，Impalad能夠註冊它們但願得到的事件類型，Statestore會週期性的發送兩種類型的消息給Impalad進程，一種爲該Impalad註冊監聽的事件的更新，基於版本的增量更新（只通知上次成功更新以後的變化）能夠減少每次通訊的消息大小；另外一種消息爲心跳信息，StateStore負責統計每個Impalad進程的狀態，Impalad能夠據此瞭解其他Impalad進程的狀態，用於判斷分配查詢任務到哪些節點。因爲週期性的推送而且每個節點的推送頻率不一致可能會致使每個Impalad進程得到的狀態不一致，因爲每一次查詢只依賴於協調者Impalad進程獲取的狀態進行任務的分配，而不須要多個進程進行再次的協調，所以並不須要保證全部的Impalad狀態是一致的。另外，StateStore進程是單點的，而且不會持久化任何數據到磁盤，若是服務掛掉，Impalad則依賴於上一次得到元數據狀態進行任務分配，官方並無提供可靠性部署的方案，一般可使用DNS方式綁定多個服務以應對單個服務掛掉的狀況。網絡

Impalad模塊

從Impalad的各個模塊能夠看出，主要查詢處理都是在Impalad進程中完成，StateStore和Catalog幫助Impalad完成元數據的管理和負載監控等工做，其實更進一步能夠將Query Planner和Query Coordinator模塊從Impalad移出單獨的做爲一個入口服務存在，而Impalad僅負責數據讀寫和子任務的執行。

在Impalad進行執行優化的時候根本原則是儘量的數據本地讀取，減小網絡通訊，畢竟在不考慮內存緩存數據的狀況下，從遠端讀取數據須要磁盤->內存->網卡->本地網卡->本地內存的過程，而從本地讀取數據僅須要本地磁盤->本地內存的過程，能夠看出，在相同的硬件結構下，讀取其餘節點數據始終本地磁盤的數據讀取速度。

Impalad服務由三個模塊組成：Query Planner、Query Coordinator和Query Executor，前兩個模塊組成前端，負責接收SQL查詢請求，解析SQL並轉換成執行計劃，交由後端執行，語法方面它既支持基本的操做（select、project、join、group by、filter、order by、limit等），也支持關聯子查詢和非關聯子查詢，支持各類outer-join和窗口函數，這部分按照通用的解析流程分爲查詢解析->語法分析->查詢優化，最終生成物理執行計劃。對於Query Planner而言，它生成物理執行計劃的過程分紅兩步，首先生成單節點執行計劃，而後再根據它獲得分區可並行的執行計劃。前者是根據相似於RDBMS進行執行優化的過程，決定join順序，對join執行謂詞下推，根據關係運算公式進行一些轉換等，這個執行計劃的生成過程依賴於Impala表和分區的統計信息。第二步是根據上一步生成的單節點執行計劃獲得分佈式執行計劃，可參照Dremel的執行過程。在上一步已經決定了join的順序，這一步須要決定join的策略：使用hash join仍是broadcast join，前者通常針對兩個大表，根據join鍵進行hash分區以使得相同的id散列到相同的節點上進行join，後者經過廣播整個小表到全部節點，Impala選擇的策略是依賴於網絡通訊的最小化。對於聚合操做，一般須要首先在每一個節點上執行預聚合，而後再根據聚合鍵的值進行hash將結果散列到多個節點再進行一次merge，最終在coordinator節點上進行最終的合併（只須要合併就能夠了），固然對於非group by的聚合運算，則能夠將每個節點預聚合的結果交給一個節點進行merge。sort和top-n的運算和這個相似。

下圖展現了執行select t1.n1, t2.n2, count(1) as c from t1 join t2 on t1.id = t2.id join t3 on t1.id = t3.id where t3.n3 between ‘a’ and ‘f’ group by t1.n1, t2.n2 order by c desc limit 100;查詢的執行邏輯，首先Query Planner生成單機的物理執行計劃，以下圖所示：

和大多數數據庫實現同樣，第一步生成了一個單節點的執行計劃，利用Parquet等列式存儲，能夠在SCAN操做的時候只讀取須要的列，而且能夠將謂詞下推到SCAN中，大大下降數據讀取。而後執行join、aggregation、sort和limit等操做，這樣的執行計劃須要再轉換成分佈式執行計劃，以下圖。

這類的查詢執行流程相似於Dremel，首先根據三個表的大小權衡使用的join方式，這裏T1和T2使用hash join，此時須要按照id的值分別將T1和T2分散到不一樣的Impalad進程，可是相同的id會散列到相同的Impalad進程，這樣每個join以後是所有數據的一部分。對於T3的join使用boardcast的方式，每個節點都會收到T3的所有數據（只須要id列），在執行完join以後能夠根據group by執行本地的預聚合，每個節點的預聚合結果只是最終結果的一部分（不一樣的節點可能存在相同的group by的值），須要再進行一次全局的聚合，而全局的聚合一樣須要並行，則根據聚合列進行hash分散到不一樣的節點執行merge運算（其實仍然是一次聚合運算），通常狀況下爲了較少數據的網絡傳輸， intermediate節點一樣也是worker節點。經過本次的聚合，相同的key只存在於一個節點，而後對於每個節點進行排序和TopN計算，最終將每個Worker的結果返回給coordinator進行合併、排序、limit計算，返回結果給用戶。

Impalad優化

上面介紹了整個查詢大體的執行流程，Impalad的後端使用的是C++實現的，這使得它能夠針對硬件作一些特殊的優化，而且能夠比使用JAVA實現的SQL引擎有更好的資源使用率。另外，後端的實現使用了LLVM，它是一個編譯器框架，能夠在執行器生成並編譯代碼。官方測試發現使用動態生成代碼機制可使得後端執行性能提升1—5倍。

在數據訪問方面，Impalad並無使用通用的HDFS讀取數據那一套流程，畢竟Impalad通常部署在DataNode上，訪問數據徹底不須要再走NameNode了，所以它使用了HDFS提供的Short-Circuit Local Reads機制，它提供了直接訪問DataNode的方案，能夠參考Hadoop官方文檔和HDFS-347瞭解詳情。

最後Impalad後端支持對中文件格式和壓縮數據的讀取，包括Avro、RC、Sequence、Parquet，支持snappy、gzip、bz2等壓縮，看來Impala不支持可能也不打算支持ORC格式啦，畢竟有自家主推的Parquet，而ORC則在Presto中普遍使用。關於Parquet和ORC等列式存儲格式可參考這裏，這裏，還有這裏。

部署方式

一般狀況下，咱們會考慮兩種方式的集羣部署：混合部署和獨立部署，下圖分別展現了混合部署與獨立部署時的各節點結構。混合部署意味着將Impala集羣部署在Hadoop集羣之上，共享整個Hadoop集羣的資源；獨立部署則是單獨使用部分機器只部署HDFS和Impala，前者的優點是Impala能夠和Hadoop集羣共享數據，不須要進行數據的拷貝，可是存在Impala和Hadoop集羣搶佔資源的狀況，進而可能影響Impala的查詢性能（MR任務也可能被Impala影響），然後者能夠提供穩定的高性能，可是須要持續的從Hadoop集羣拷貝數據到Impala集羣上，增長了ETL的複雜度。兩種方式各有優劣，可是針對前一種部署方案，須要考慮如何分配資源的問題，首先在混合部署的狀況下不可能再讓Impalad進程常駐（這樣至關於把每個NodeManager的資源分出去了一部分，而且不能充分利用集羣資源），可是YARN的資源分配機制延遲太大，對於Impala的查詢速度有很大的影響，因而Impala很早就設計了一種在YARN上完成Impala資源調度的方案——Llama（Low Latency Application MAster），它實際上是一個AM的角色，對於Impala而言。它的要求是在查詢執行以前必須確保須要的資源可用，不然可能出現一個Impalad的阻塞而影響整個查詢的響應速度（木桶原理），Llama會在Impala查詢以前申請足夠的資源，而且在查詢完成以後儘量的緩存資源，只有當YARN須要將該部分資源用於其它工做時，Llama纔會將資源釋放。雖然Llama儘量的保持資源，可是當混合部署的狀況下，仍是可能存在Impala查詢獲取不到資源的狀況，因此爲了保證高性能，仍是建議獨立部署。

測試

咱們小組的同事對Impala作了一次基於TPCDS數據集的性能測試，分別基於1TB和10TB的數據集，能夠看出，它的查詢性能較之於Hive有數量級級別的提高，對比Spark SQL也有幾倍的提高，Compute stat操做能夠給Impala帶來必定的查詢優化，可是偶爾反而誤導查詢優化器以致於性能降低，最後咱們還測試了Impala on Kudu，發現它並無達到意料中的性能（幾倍的差異）。惟一的缺憾是咱們並無對多用戶併發場景下進行測試，不過從單個查詢的資源消耗來看，C++實現的Impala對資源的消耗也是最少的，能夠推斷出在多用戶下它仍然能知足快速響應的需求，最後是官方給出的多用戶場景下的對比結果（有點故意黑Presto的感受）。

1TB數據集與spark對比測試結果

10TB數據集與spark對比測試結果

Impala on parquet與Impala on Kudu對比測試結果

併發測試結果

總結

本文主要介紹了Impala這個高性能的ad-hoc查詢引擎，分別從使用、原理和部署等方面作了詳細的分析，最終基於咱們的測試結果也證明了它的高性能，區別於傳統DBMS的MPP解決方案，例如Greenplum、Vertica、Teradata等，Impala更好的融入大數據（Hadoop/Spark）生態圈，更好的實現數據之間的流通，而傳統MPP數據庫，更傾向於數據自制。固然基於HDFS的實現致使Impala沒法實現單條數據的實時更新，而只能批量的追加或者覆蓋數據，雖然Cloudera也提供了Impala對於Kudu的支持，可是從性能測試結果看，目前查詢性能仍是不理想，而傳統MPP數據庫不只能夠支持單條數據的實時更新，甚至可以在保證查詢性能的狀況下支持較複雜的事務，這也是SQL-on-Hadoop查詢引擎所可望不可即的。可是不管如何，這類的查詢引擎畢竟支持SQL引擎而不是一個完整的數據庫系統，它提供給用戶在大數據圈中高性能的查詢服務，這也可以知足了大部分用戶的需求。