面向ad-hoc查询的实时SQL分析系统

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查询优化器。

来源于：http://yanbohappy.sinaapp.com/?p=220

秒客网