数据分析查询引擎Impala初探

Impala是 Cloudera 公司主导开发的新型查询系统，它提供 SQL 语义，能查询存储在 Hadoop 的 HDFS 和 HBase 中的 PB 级大数据。已有的 Hive 系统虽然也提供了 SQL 语义，但由于 Hive 底层执行使用的是 MapReduce 引擎，仍然是一个批处理过程，难以满足查询的交互性。相比之下，Impala 的最大特点也是最大卖点就是它的快速。

在介绍 Impala 之前需要先介绍 Google 的 Dremel 系统，因为 Impala 最开始是参照 Dremel 系统进行设计的。Dremel 是 Google 的交互式数据分析系统，它构建于 Google 的 GFS（Google FileSystem）等系统之上，支撑了 Google 的数据分析服务 BigQuery 等诸多服务。Dremel 的技术亮点主要有两个：一是实现了嵌套型数据的列存储；二是使用了多层查询树，使得任务可以在数千个节点上并行执行和聚合结果。列存储在关系型数据库中并不陌生，它可以减少查询时处理的数据量，有效提升查询效率。Dremel 的列存储的不同之处在于它针对的并不是传统的关系数据，而是嵌套结构的数据。Dremel 可以将一条条的嵌套结构的记录转换成列存储形式，查询时根据查询条件读取需要的列，然后进行条件过滤，输出时再将列组装成嵌套结构的记录输出，记录的正向和反向转换都通过高效的状态机实现。另外，Dremel 的多层查询树则借鉴了分布式搜索引擎的设计，查询树的根节点负责接收查询，并将查询分发到下一层节点，底层节点负责具体的数据读取和查询执行，然后将结果返回上层节点。

Impala 在受到 Google 的 Dremel 启发下开发的实时交互 SQL 大数据查询工具，Impala 没有再使用缓慢的 Hive+MapReduce 批处理，而是通过使用与商用并行关系数据库中类似的分布式查询引擎（由 QueryPlanner、QueryCoordinator 和 QueryExecEngine 三部分组成），可以直接从 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 主要包含 QueryPlanner、QueryCoordinator 和 QueryExecEngine 三个模块。QueryPalnner 接收来自 SQLAPP 和 ODBC 的查询，然后将查询转换为许多子查询，QueryCoordinator 将这些子查询分发到各个节点上，由各个节点上的 QueryExecEngine 负责子查询的执行，最后返回子查询的结果，这些中间结果经过聚集之后最终返回给用户。

Impala 的组成

Impala 主要由 Impalad，StateStore 和 CLI 组成。

Impalad

与 DataNode 运行在同一节点上，由 Impalad 进程表示，它接收客户端的查询请求（接收查询请求的 Impalad 为 Coordinator，Coordinator 通过 JNI 调用 java 前端解释 SQL 查询语句，生成查询计划树，再通过调度器把执行计划分发给具有相应数据的其它 Impalad 进行执行），读写数据，并行执行查询，并把结果通过网络流式的传送回给 Coordinator，由 Coordinator 返回给客户端。同时 Impalad 也与 StateStore 保持连接，用于确定哪个 Impalad 是健康和可以接受新的工作。在 Impalad 中启动三个 ThriftServer:beeswax_server（连接客户端），hs2_server（借用 Hive 元数据），be_server（Impalad 内部使用）和一个 ImpalaServer 服务。每个 impalad 实例会接收、规划并调节来自 ODBC 或 ImpalaShell 等客户端的查询。每个 impalad 实例会充当一个 Worker，处理由其它 impalad 实例分发出来的查询片段(query fragments)。客户端可以随便连接到任意一个 impalad 实例，被连接的 impalad 实例将充当本次查询的协调者（Ordinator)，将查询分发给集群内的其它 impalad 实例进行并行计算。当所有计算完毕时，其它各个 impalad 实例将会把各自的计算结果发送给充当 Ordinator 的 impalad 实例，由这个 Ordinator 实例把结果返回给客户端。每个 impalad 进程可以处理多个并发请求。

Impala StateStore

跟踪集群中的 Impalad 的健康状态及位置信息，由 statestored 进程表示，它通过创建多个线程来处理 Impalad 的注册订阅和与各 Impalad 保持心跳连接，各 Impalad 都会缓存一份 StateStore 中的信息，当 StateStore 离线后（Impalad 发现 StateStore 处于离线时，会进入 recovery 模式，反复注册，当 StateStore 重新加入集群后，自动恢复正常，更新缓存数据）因为 Impalad 有 StateStore 的缓存仍然可以工作，但会因为有些 Impalad 失效了，而已缓存数据无法更新，导致把执行计划分配给了失效的 Impalad，导致查询失败。

• 用于协调各个运行 impalad 的实例之间的信息关系，Impala 正是通过这些信息去定位查询请求所要的数据。换句话说，statestore 的作用主要为跟踪各个 impalad 实例的位置和状态，让各个 impalad 实例以集群的方式运行起来。
• 与 HDFS 的 NameNode 不一样，虽然 StateStore 一般只安装一份，但一旦 StateStore 挂掉了，各个 impalad 实例却仍然会保持集群的方式处理查询请求，只是无法将各自的状态更新到 StateStore 中，如果这个时候新加入一个 impalad 实例，则新加入的 impalad 实例不为现有集群中的其他 impalad 实例所识别（事实上，经笔者测试，如果 impalad 启动在 statestored 之后，根本无法正常启动，因为 impalad 启动时是需要指定 statestored 的主机信息的）。然而，StateStore 一旦重启，则所有 StateStore 所服务的各个 impalad 实例（包括 statestore 挂掉期间新加入的 impalad 实例）的信息（由 impalad 实例发给 statestore）都会进行重建。

CLI（Impala shell）

提供给用户查询使用的命令行工具（ImpalaShell 使用 python 实现），同时 Impala 还提供了 Hue，JDBC，ODBC 使用接口。该客户端工具提供一个交互接口，供使用者发起数据查询或管理任务，比如连接到 impalad。这些查询请求会传给 ODBC 这个标准查询接口。说白了，就是一个命令行客户端。

Impala 的功能特性

查询执行

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）mergesort得到最终topN结果。

Backend从frontend接收plan segment并执行，执行性能非常关键，impala采取的查询性能优化措施有：

• 向量执行。一次getNext处理一批记录，多个操作符可以做pipeline。
• LLVM编译执行，CPU密集型查询效率提升5倍以上。
• IO本地化。利用HDFS short-circuit local read功能，实现本地文件读取
• Parquet列存，相比其他格式性能最高提升5倍。

资源管理

impala通常与MR等离线任务运行在一个集群上，通过YARN统一管理资源，如何同时满足交互式查询和离线查询两种需求具有较大挑战性。YARN通过全局唯一的ResourceMananger调度资源，好处是RM拥有整个集群全局信息，能做出更好调度决策，缺点是资源分配的性能不足。Impala每个查询都需要分配资源，当每秒查询数上千时，YARN资源分配的响应时间变的很长，影响到查询性能。目前通过两个措施解决这个问题：

• 引入快速、非集中式的查询准入机制，控制查询并发度。
• LLAM（low latency application master）通过缓存资源，批量分配，增量分配等方式实现降低资源分配延时

相对于Hive所使用的优化技术：

• 没有使用MapReduce进行并行计算，虽然MapReduce是非常好的并行计算框架，但它更多的面向批处理模式，而不是面向交互式的SQL执行。与MapReduce相比：Impala把整个查询分成一执行计划树，而不是一连串的MapReduce任务，在分发执行计划后，Impala使用拉式获取数据的方式获取结果，把结果数据组成按执行树流式传递汇集，减少的了把中间结果写入磁盘的步骤，再从磁盘读取数据的开销。Impala使用服务的方式避免每次执行查询都需要启动的开销，即相比Hive没了MapReduce启动时间。
• 使用LLVM产生运行代码，针对特定查询生成特定代码，同时使用Inline的方式减少函数调用的开销，加快执行效率。
• 充分利用可用的硬件指令（2）。
• 更好的IO调度，Impala知道数据块所在的磁盘位置能够更好的利用多磁盘的优势，同时Impala支持直接数据块读取和本地代码计算checksum。
• 通过选择合适的数据存储格式可以得到最好的性能（Impala支持多种存储格式）。
• 最大使用内存，中间结果不写磁盘，及时通过网络以stream的方式传递。

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与Hive的关系

Impala与Hive都是构建在Hadoop之上的数据查询工具各有不同的侧重适应面，但从客户端使用来看Impala与Hive有很多的共同之处，如数据表元数据、ODBC/JDBC驱动、SQL语法、灵活的文件格式、存储资源池等。Impala与Hive在Hadoop中的关系下图所示。Hive适合于长时间的批处理查询分析，而Impala适合于实时交互式SQL查询，Impala给数据分析人员提供了快速实验、验证想法的大数据分析工具。可以先使用hive进行数据转换处理，之后使用Impala在Hive处理后的结果数据集上进行快速的数据分析。

Impala支持SQL92中的大部分select语句，以及SQL2003标准中的分析函数。不支持DELETE和UPDATE，但是支持批量装载数据（insert into select, LOAD DATA)和批量删除数据（drop partition）。除此之外，用户也可直接操作HDFS文件实现数据装载和清理。

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: 在遇到内存放不下数据时，当前版本 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 代替执行，不会影响服务。对于 StateStore 目前只有一个，但当 StateStore 失效，也不会影响服务，每个 Impalad 都缓存了 StateStore 的信息，只是不能再更新集群状态，有可能会把执行任务分配给已经失效的 Impalad 执行，导致本次 Query 失败。
• 适用面：
  • Hive: 复杂的批处理查询任务，数据转换任务。
  • Impala：实时数据分析，因为不支持 UDF，能处理的问题域有一定的限制，与 Hive 配合使用, 对 Hive 的结果数据集进行实时分析。

Impala 与 Shark，Drill 等的比较

开源组织 Apache 也发起了名为 Drill 的项目来实现 Hadoop 上的 Dremel，目前该项目正在开发当中，相关的文档和代码还不多，可以说暂时还未对 Impala 构成足够的威胁。从 Quora 上的问答来看，Cloudera 有 7-8 名工程师全职在 Impala 项目上，而相比之下 Drill 目前的动作稍显迟钝。具体来说，截止到 2012 年 10 月底，Drill 的代码库里实现了 queryparser, planparser，及能对 JSON 格式的数据进行扫描的 planevaluator；而 Impala 同期已经有了一个比较完毕的分布式 queryexecution 引擎，并对 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 进行了一次对比试验，在 ScanQuery，AggregationQuery 和 JoinQuery 三种类型的任务中对它们进行了比较。图 2 就是 AMPLab 报告中 AggregationQuery 的性能对比。在图中我们可以看到，商业版本的 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/