大数据时代快速SQL引擎-Impala

背景

随着大数据时代的到来，Hadoop在过去几年以接近统治性的方式包揽的ETL和数据分析查询的工做，你们也无心间的想往大数据方向靠拢，即便天天数据也就几10、几百M也要放到Hadoop上做分析，只会拔苗助长，可是当面对真正的Big Data的时候，Hadoop就会暴露出它对于数据分析查询支持的弱点。甚至出现《MapReduce: 一个巨大的倒退》此类极端的吐槽，这也怪不得Hadoop，毕竟它的设计就是为了批处理，使用用MR的编程模型来实现SQL查询，性能确定不如意。因此一般我也只是把Hive当作可以提供将SQL语义转换成MR任务的工具，尤为在作ETL的时候。

在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的第一选择。

从用户视角

能够将Impala这类系统的用户分为两类，一类是负责数据导入和管理的数据开发同窗，另外一类则是执行查询的数据分析师同窗，前者一般须要将数据存储到HDFS，经过CREATE TABLE的方式建立与数据match的schema，而后经过load data或者add partition的方式将表和数据关联起来，这一些流程串起来仍是挺麻烦的，可是多亏了Hive，因为Impala能够共享Hive的MetaStore，这样就可使用Hive完成此类ETL工做，而后将数据查询的工做交给Impala，大大简化工做流程（据我所知毕竟大部分数据开发同窗仍是比较熟悉Hive）。接下来对于数据分析师而言就是如何编写正确的SQ以表达他们的查询、分析需求，这也是它们最拿手的了，Impala一般能够在TB级别的数据上提供秒级的查询速度，因此使用起来可能让你从Hive的龟速响应一下提高到指望的速度。

Impala除了支持简单类型以外，还支持String、timestamp、decimal等多种类型，用户还能够对于特殊的逻辑实现自定义函数（UDF）和自定义聚合函数（UDAF），前者可使用Java和C++实现，后者目前仅支持C++实现，除此以外的schema操做均可以在Hive上实现，因为Impala的存储由HDFS实现，所以不可以实现update、delete语句，若是有此类需求，仍是须要从新计算整个分区的数据而且覆盖老数据，这点对于修改的实时性要求比较高的需求仍是不能知足的，若是有此类需求仍是期待Kudu的支持吧，或者尝试一下传统的MPP数据库，例如GreenPlum。

当完成数据导入以后，用户须要执行COMPUTE STATS

系统架构

从用户的使用方式上来看，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引擎而不是一个完整的数据库系统，它提供给用户在大数据圈中高性能的查询服务，这也可以知足了大部分用户的需求。

参考

Impala: A Modern, Open-Source SQL Engine for Hadoop

Dremel: interactive analysis of web-scale datasets

Impala原理及其调优

Impala：新一代开源大数据分析引擎

Apache Impala Documents