大数据分析查询引擎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/