Impala/Hive現狀分析與前景展望

Impala和Hive野史

提到Impala就不得不提Google的Dremel，處理PB級數據規模的基於SQL的交互式、實時數據分析系統。Dremel是Google推出的PaaS數據分析服務BigQuery的後臺。Google已經有了MapReduce，爲何還要開發Dremel呢？Dremel/Impala類系統和MapReduce有什麼區別呢？

Hadoop如今已經成爲BigData應用系統的標配，那麼基於Hadoop平臺作大數據分析無非幾種使用方式：

	優勢	缺點	典型案例
本身寫MapReduce任務	性能比Hive和Pig要高點	開發難度大	1） 搜索引擎網頁處理，PageRank計算（Google）2） 典型的ETL（全盤掃描）3） 機器學習/聚類，分類，推薦等（百度Ecomm）
使用Hive作基於SQL的分析	對於數據分析師來講SQL太熟悉了	有些場景下性能不如MR	1） 用戶訪問日誌處理/互聯網廣告（Yahoo, Facebook, hulu, Amazon）2） 電子商務（淘寶的雲梯）
使用Pig作數據分析	Pig的語法不是很普及	有些場景下性能不如MR	統計和機器學習（Yahoo, twitter）
基於HBase開發的系統	基本能夠達到準實時統計分析功能	目前沒有開源實現，開發成本高	大可能是自有系統，例如Google的Percolator，淘寶的prom

關於twitter使用Pig作機器學習方面的內容請參考SIGMOD2012的論文Large-Scale Machine Learning at Twitter。

咱們都知道MapReduce是由Google發明的，Google發明這個固然首先是知足本身的應用需求。它們的主要需求就是對互聯網網頁的處理：網頁有效信息提取，轉化，PageRank的計算。這種應用模式決定了這是一個批處理的系統。後來Facebook爲了瞭解用戶對其平臺上廣告點擊的反饋，同時給不會MR編程只會使用SQL的數據分析師開發了Hive這個東西，使得Hive在FB內部應用很是普遍。Yahoo設計Hadoop的時候是想把Pig提拔成Hadoop平臺的SQL標準，沒想到半路殺出個程咬金，並且在Hive在工業界反響至關好，使得Hive的使用很是普及。目前互聯網公司最主要的盈利模式是廣告，基於用戶訪問日誌分析提升用戶的廣告點擊率，我認爲這個典型應用是目前Hadoop應用中最主要的應用場景。怪不得Jeff Hammerbacher認爲把他主要的心思放到讓用戶去點擊廣告上是件愚蠢的事情，這哥們後來跳到Cloudera當了首席科學家。

可見Hive確實很是普及，國內的互聯網公司也大多數在用Hive。可是Hive有個很大的缺點就是太慢了，面向的是批處理。不少問題是有時效性的，數據一旦過了時效窗口就失去了意義。因此在大數據領域很是須要一個面向interactive，面向ad-hoc查詢的實時SQL分析系統，在Dremel的啓發下，Impala誕生了。

Impala能夠認爲是在大數據領域的MPP，因此不少地方是很像Greenplum, AsterData這樣的商用數據倉庫產品的。因此當年MapReduce與MPP之爭也算是有了個結果。Impala和這些商用系統的最大區別就是：Impala的可擴展性更好，支持的規模更大，面向的底層存儲和硬件系統是commodity hardware。

Impala設計目標

分佈式環境下通用SQL引擎：既支持OLTP也支持OLAP

SQL查詢的規模和粒度：從毫秒級到小時級

底層存儲依賴HDFS和Hbase

使用更加高效的C++編寫

SQL的執行引擎借鑑了分佈式數據庫MPP的思想而再也不依賴MapReduce

Impala系統架構

1， SQL Interface

目前這部分是借用Hive的，包括ODBC/Beeswax。Client的SQL查詢經過ODBC/Beeswax的Thrift API發送到集羣內部的任何一個impalad，而後這個impalad就成了這個query的coordinator。

2， Unified metastore

Impala中表的元數據存儲借用的是Hive的，也就是用個RDBMS來存儲Impala中表的元數據信息。Impala本身提供一個叫statestored的進程負責收集分佈在集羣中各個impalad進程的資源信息，用於query的調度（這個功能會在2013Q1末GA版本會提供）。Statestored對外提供Thrift服務。這個statestored未來還會有個功能就是把impala表的metadata分發到各個impalad中（也是在2013Q1末GA版本中提供）。

3， Impala daemon

名爲impalad的進程，主要有兩個角色：一是協調client提交的query的執行，給其餘impalad分配任務，收集其餘impalad的執行結果進行彙總；二是這個impalad也會執行其餘impalad給其分配的任務，在執行這部分任務主要就是對本地HDFS和HBase裏的部分數據進行操做了（都是本地IO操做，HDFS還支持dfs.client.read.shortcircuit跨過網卡直接磁盤讀）。

Impala系統優缺點

目前支持Hive SQL的大部分功能，例如select, insert, where, join, union, subqueries, aggregation, order by only with limit。

Trevni文件格式是一個性能提高的突破點。

DDL經過Hive操控Hive的metastore來完成，由於Impala使用了Hive的metastore。

侷限性：不支持UDF，不支持SerDes，只支持in-memory join，只有基本的cost-based optimizer。

Query執行過程

1,用戶經過ODBC/Beeswax Thrift API提交query到某個impalad。Impalad的Query Planner使用jflex和CUP解析SQL語句。而後Planner把這個query的parse trees變成若干PlanFragment，而後把PlanFragment發送到backend/Query Coordinator。

PlanFragment由PlanNode組成的，能被分發到單獨節點上原子執行，每一個PlanNode表示一個relational operator和對其執行優化須要的信息。例如：AggregationNode, ExchangeNode, HBaseScanNode, HashJoinNode, HdfsScanNode, MergeNode, SortNode

2,Coordinator初始化相應impalad上的任務執行（存儲了這個query相關數據的節點都會被分配任務）。

3,Query Executor經過流式交換中間輸出。Query Coordinator彙總來自各個impalad的結果後返回給client；

在執行過程當中若是遇到聚合函數limit n時，能夠直接在每一個impalad上截取top-n（該功能也是在2013Q1末GA版本提供）。

對於distributed-aggregation，仍是先在各個impalad上作局部aggregation，而後在coordinator節點上merge aggregation。目前貌似這個功能還作的很弱，基本上至關於reduce=1的MapReduce join。據說更強大的hash-partitioned aggregation/partitioned join正在開發中，這個feature很期待啊。

下面以一個SQL語句的執行過程爲例說明：這個例子來自 http://www.sizeofvoid.net/wp-content/uploads/ImpalaIntroduction2.pdf  「SQL breakdown sample」一節裏的例子，這是個查詢，有JOIN，有條件查詢，有aggregation，有sort的例子，基本上啥都有了。

select i_item_id, i_list_price, avg(ss_sales_price) agg1
FROM store_sales
JOIN item on (store_sales.ss_item_id = item.i_item_id)
JOIN customer on (store_sales.ss_customer_id = customer.c_id)
Where
i_list_price > 1000 and
c_gender = ‘M’ and
c_marital_status = ‘S‘ and
c_city in (‘Beijing’,'Shanghai’,'Guangzhou’)
group by i_item_id,
order by i_list_price
limit 1000
生成的plan tree是這樣的：

並且還標明瞭哪些是能夠分佈式執行的，哪些是不能分佈式執行的。

JOIN是數據庫最重要的問題之一，通常的實現方法主要有Nested Loop Join，Sort-Merge Join和Hash Join。通常來講，查詢優化器會首先考慮Nested Loop和Sort-Merge，但若是兩個表都比較大且沒有合適的索引時，纔會考慮使用Hash Join。通常狀況下只有Nested Loop Join能用在非等值join裏。 關於數據庫中的JOIN算法能夠參考這篇文章：http://www.mysqlops.com/2011/03/03/db-join-algorithm.html 。

那麼在Hadoop中JOIN是怎麼實現的呢？Hadoop作join主要有reduce-side join和map-side join兩種方式。map-side join又能夠分爲小表所有載入內存和小表分塊載入內存(bucket join）兩種方式。 有關Hadoop join算法的實現能夠參考這篇文章： http://dongxicheng.org/mapreduce/hadoop-join-two-tables/

那麼Impala的JOIN是怎麼作的？目前仍是in-memory hash join，也就是參與JOIN的表一大一小，把那小表所有讀到內存裏。淘寶的MyFox系統當時也是這麼作的，記得去年參加淘寶的校園招聘宣講會的時候我還問了這個問題呢。Impala已經在開發partitioned hash join了，不知道2013Q1末咱們能不能用上啊。

性能測試

目前我已知的有兩份測試數據：

這是Intel的測試數據，4節點集羣比較了shark, impala和hive的性能：

• count – counting the entire rows of the table;
• groupby – find the sum of a column grouped by a key for the input table and limit result rows to 1;
• join – join two input tables on specified keys and limit result rows to 1.

https://groups.google.com/forum/?fromgroups=#!topic/shark-users/IJ1U056dhDI

另外一份來自http://www.sizeofvoid.net/wp-content/uploads/ImpalaIntroduction2.pdf ：

Up to 90˟ times faster, compared with Hive

• Purely I/O bound scenario, 3-4˟
• with joins, 7-45˟
• with memory cached, 20-90˟

代碼結構

主要分爲be/backend和fe/frontend，各部分功能以下：

最後，我想說一句的是，即便有了Impala，MapReduce在ETL方面仍是有用的。

Impala目前bug太多，還不能用於工業生產，若是沒有跳票的話，2013Q1末會有GA，期待那個時候會有穩定版本可用。

Hive/Impala暢想

Metadata除了存儲表格元數據之外還應該存儲一些表格的統計信息用來作SQL代價估計和執行優化。例如每一列數據分佈的柱狀圖。

大表JOIN和group by在Impala裏也是很是有挑戰的issue。

Bucket join/partition join應該是Impala下一步很是迫切的需求。

既然Impala會把HBase做爲底層存儲，而廣泛意義上認爲HBase是爲了寫優化而設計的。那麼HBase的讀優化也是一個要提到日程的話題。

Avro, RCFile, Trevni（列存儲和輕量級壓縮）和Impala的融合是個很是值得期待的話題。

大表JOIN算法猜測：生成join中間表，不過這個表只有少數的幾列：primary key 和 join column，這個中間表就很是小了，節省了不少網絡傳輸帶寬。而後就把它弄到各個Region上過濾知足條件的Record。

參考文獻：

http://www.sizeofvoid.net/wp-content/uploads/ImpalaIntroduction2.pdf

http://www.slideshare.net/ChicagoHUG/an-introduction-to-impala-low-latency-queries-for-apache-hadoop

在Impala研究分析過程當中獲得了個人師弟 @徐偉辰雲計算  的不少幫助，他也是一個對雲計算、大數據很是感興趣的童鞋 。咱們正在合做設計和開發一個基於Hive/Impala的SQL查詢優化器。