Impala 在 Hulu 中的优化和改进

点击hadoop123 关注我哟

知名的大数据中台技术分享基地，涉及 大数据架构 (hadoop/spark/flink等) ， 数据平台(数据交换、数据服务、数据治理等) 和 数据产品(BI、AB测试平台) 等，也会分享 最新技术进展，大数据相关职位和求职信息，大数据技术交流聚会、讲座以及会议 等。

背景

Impala 是一个 SQL on Hadoop 的 MPP 查询引擎，由 Cloudera 主导开发并捐献给 Apache 软件基金会，在 2017 年底正式孵化成为 Apache 顶级项目。

如图，大数据领域里的 OLAP 系统种类繁多，它们各自有自己的特长和局限性，需要根据实际业务场景合理进行选择。存储方式往往决定了系统的能力和兼容性，跟据存储方式我们可以把它们分为三大类：

1. 不存储原始数据，只保留聚合结果，如 Druid 、 Kylin 等

2. 存储计算一体，自己实现存储层，如 Greenplum 、 Clickhouse 、 Doris （原 Baidu Palo ）等

3. 存储计算分离，依托 HDFS 、 S3 等实现存储层，这类系统又可以细分为两类：

1. 使用自定义的文件格式，如 HAWQ 、 VectorH 等

2. 使用开源文件格式，如 Hive 、 Impala 、 Presto 、 SparkSQL 、 Drill 等

第 1 类系统不保留原始数据，可以把性能做得很高，但由于聚合方式需要事先定义，比较适合报表类业务等查询模式相对固定的场景。当需要明细查询或交互式的灵活查询（ Adhoc 查询）时，仍需要另两类系统的加入才能支持。

第 2 类系统自己管理数据，可以做很多定制的优化，从而也能达到较高的性能。但由于数据没法与其它系统共享，往往需要将已有数据重新导入一次，在引入时需要考虑迁移成本以及多存一份数据的代价，因此一般在新建数仓中被考虑使用。第 3.a 类系统虽然依托外部存储，但因为采用封闭的文件格式，仍需要多存一份数据，其地位其实与第 2 类相同，常常在新建数仓时才被考虑。

使用最广的还是属于 3.b 类的系统，主要原因是基于 Hadoop 的数仓使用广泛，此类系统不需要重新生成一份数据，能与已有架构充分兼容。在此类系统中， Impala 、 Presto 、 Drill 属于 MPP （ Massively Parallel Processing ）系统，各节点流式地在内存中完成计算，中间数据几乎不落盘，相比 Hive 、 SparkSQL 等批处理系统能达到极高的性能。

在 Hadoop 生态圈的 MPP 系统中， Impala 具有优异的性能，它的优势缘于：

• C++ 实现，相比 Presto 、 Drill 等 Java 实现要更高效，也省去了 Java GC 的开销

• 基于 LLVM 的 Code Generation ，能根据实际数据类型生成高效的执行代码

• CBO (Cost Based Optimizer) ，基于各表的统计信息得出代价最低的执行计划

• 缓存元数据，生成执行计划时不再需要和 Hive 、 HDFS 进行交互

• RuntimeFilter ，在运行时基于已读的小表数据裁剪大表需要扫描的数据量

• Predicate/Aggregation Pushdown ，虽然 MPP 系统都会有下推优化，但支持的程度各有不同，也依赖于底层文件格式的支持。 Impala 在 Parquet 存储格式上做了很多 native 的下推优化，如 IMPALA-3654 、 IMPALA-4624 、 IMPALA-4985 、 IMPALA-6113 等。

Impala在Hulu的数仓中 有很多应用 ，我们对 Impala 做了一些内核级别的开发和优化，希望能与大家共同探讨。

Impala内部原理

Impala 集群由一台 Catalog Server ( 简称 catalogd) ，一台 Statestore Server ( 简称 statestore) 和若干 Impala Daemon ( 简称 impalad) 组成。自 Impala 2.10 之后， Impalad 又可分为 Coordinator 或 Executor 两种角色。各服务的功能如下：

• Statestore 负责管理集群心跳和广播元数据更新。

• Catalogd 负责从 Hive 和 HDFS 拉取元数据并缓存下来，同时将元数据更新发给 statestore 进行广播。另外 catalogd 还负责执行建表、新建 partition 等 DDL 语句。

• Impalad 中的 Coordinator 也会缓存元数据，负责接收 SQL 查询请求并生成执行计划，并将执行计划的分片（ PlanFragment ）调度到各 Executor 去执行，最终汇总结果返回给客户端。

• Impalad 中的 Executor 则只负责 PlanFragment 的执行。

下图展示了一个 SQL 查询的执行过程：

(1) 只有 Coordinator 角色的 Impalad 才会接收客户端请求。 Coordinator 对查询进行语法分析、语义分析。

(2) 语义分析中需要查询各表的元数据（元数据的具体内容见后文），如果在该 Coordinator 的元数据缓存中缺失，则会向 catalogd 请求加载。 catalogd 会向 Hive Metastore 和 HDFS NameNode 查询所要的元数据，并将元数据的更新发送给 statestore 进行广播，从而所有 Coordinator 都会得到更新。

(3) Coordinator 通过语义分析生成执行计划，并根据数据的本地性（ locality ）将 Plan Fragment 调度到各 Executor 去执行

(4) Executor 从 HDFS 读取数据，并将 PlanFragment 实例的执行结果返回给上层结点，最终汇总到 Coordinator 得到最终结果。

Impala 的执行计划遵从 Volcano 的 Iterator 模型，是由若干 PlanNode 组成的执行计划树，叶子节点读取外部数据并传输给上层节点做下一步处理，最终在根节点汇总。

Iterator 模型使得整个查询可以最大限度地流式进行，从而降低了查询的延迟。 Volcano 模型的另一大贡献是引入了 Exchange 节点，使得执行计划可以划分为不同的分片，各自采用合适的并行度去执行。

如图是 TPC-H Benchmark 中 Query3 的执行计划，根据是否需要 broadcast 、 shuffle 等被切分成几个 PlanFragment 。每个 Executor 执行的就是 Plan Fragment 的实例。

Hulu对Impala的改进

1. 增加对ORC文件格式的支持

ORC 是一种列式存储的文件格式，由 Hortonworks 主导开发，而 Cloudera 主推的是 Parquet 。因此在 Cloudera 和 Hortonworks 宣布合并之前， Impala 并没有支持 ORC 的计划。由于历史原因， Hulu 的 Hive 数仓中大量使用了 ORC 存储格式，为了引入 Impala ，我们决定对它进行内核级的修改，让其支持 ORC 存储格式。我们分两步走，第一步先实现基本类型（ primitive types ）的支持（ IMPALA-5717 ），第二步再增加了嵌套类型（ struct 、 array 、 map ）的支持（ IMPALA-6503 ）。这两部分工作均已贡献给社区， impala 在 2.12 及 3.1 版本开始支持读取 ORC 文件中基本类型的列，在 3.2 版本支持读取 ORC 文件中嵌套类型的列。

这部分工作的核心是实现一个 HdfsOrcScanner ，因为 Query 执行的大部分逻辑如语法分析、语义分析、调度等基本可以复用已有的实现，唯独最终解析 ORC 文件这块需要专门的实现。前面我们介绍过了 Impala 的执行计划树，树的叶子节点都是 ScanNode 。每个 ScanNode 的实例负责读取若干个数据分片（ split ），每个 split 由一个 Scanner 线程去处理。如下图所示，编号 02 的 HdfsScanNode 有 14 个实例，分别运行在 14 个 Executor 上。每个实例会启动若干个 Scanner 线程来读取 split 。

Impala 支持的所有 HDFS 文件格式（如 Parquet 、 Avro 、 SequenceFile 、 RCFile 、 Text 等）都有一个对应的 scanner 实现，为了支持 ORC ，我们同样要实现一个 HdfsOrcScanner 。

上图是 HdfsOrcScanner 的内部结构，主要可分为以下几方面：

(1) Impala 如何管理内存 ： Impalad 会追踪每个查询占用了自己多少内存，超过阈值的查询会被 kill 掉。 HdfsOrcScanner 的内存管理要遵从既有的流程，从而让 impalad 能正确统计内存占用量（通过 impala::MemTracker ）。

(2) Impala 如何读取数据 ： Scanner 并不需要真正读取 HDFS 上的数据， Impala 把 IO 读取封装成了 DiskIoMgr 。 ORC 文件的读取并不是从头读到尾，而是先解析文件尾得到元信息，然后跳到每个 Stripe （行组）中读取所需的列。每个 Stripe 的读取又要先解析 Stripe 尾部的元信息。这些都要求 Scanner 正确地与 DiskIoMgr 进行交互，避免无用的 IO 。

(3) Impala 如何表示数据 ：不管底层文件是列存还是行存， Scanner 都会将其物化（ materialization ）成为内存中按行存放的 Tuple ，若干个 Tuple 组成 RowBatch 返回给 ScanNode 。每个 Tuple 包含了一行中被选择的各列数据，具体的样子由 TupleDescriptor 进行描述。 Scanner 需要理解 TupleDescriptor ，并将 ORC 数据物化成所需的 Tuple 。这块的工作比较细，比如需要考虑 Tuple 所引用的内存空间的生命周期管理、 TupleDescriptor 所要的列在 ORC 文件中是否存在及是否兼容、遇到正常中断（如被 cancel 或达到 limit ）或解析异常时的处理等。

(4) 如何解析 ORC 格式的文件 ： ORC Reader 已经有 C++ 版的官方开源实现，我们直接将其封装在 HdfsOrcScanner 里即可，主要的工作是把前 3 个层面封装成 ORC Reader 的参数或输入，并解析 ORC Reader 的输出。在集成 ORC Reader （属于 ORC library ）的过程中，我们还发现并修复了一些 bug ，详见 ORC-311 、 ORC-312 、 ORC-313 、 ORC-314 、 ORC-317 、 ORC-403 。

上图对比了 Impala on Parquet 、 Impala on ORC 、 Presto on ORC在TPC-H基准测试中的20个查询的耗时（单位：秒）。可以看到Impala on ORC的性能虽然比不上Impala on Parquet，但相比Presto on ORC还是有很大的优势。 Impala的ORC scanner还有很多优化可做，比如支持Aggregation Pushdown、结合ORC文件的统计信息来减少无用IO、使用DiskIoMgr的异步IO接口等，理论上应该能达到与Parquet scanner相近的性能。关于后续的工作，欢迎关注IMPALA-6505、IMPALA-6636、IMPALA-8046等相关JIRA。

2. 自动刷新元数据

Impala 缓存了 Hive 中各表的元数据，包括列的定义、 partition 的位置和权限、 HDFS 文件的信息（大小、权限、复本位置等）。这些信息从 Hive Metastore (HMS)  和 HDFS NameNode (NN) 得来，当查询再次访问相同的表时， Impala 可以利用缓存的元数据直接生成执行计划并开始执行，省去了对 HMS 和 NN  的多次交互。

这是 Impala 元数据层的设计初衷，确实加速了查询性能，也降低了对 HMS 和 NN 的访问压力，但因此引入了两个非常不友好的语句:  "INVALIDATE METADATA"  和  "REFRESH" 。

当 Hive 中的表有更新时（如新增 partition 或重新覆盖了原表数据）， Impala 并不能自动感知，需要用户手动执行 " REFRESH tableName"  语句来刷新元数据缓存。如果在 Hive 中建了一个新表，还需要在 Impala 中执行 " INVALIDATE METADATA tableName"  来通知 Impala 这个新表的存在。如果没有及时操作，对应表上的查询基本都会挂掉。

为了将 Impala 无缝引入我们已有的 Hadoop 数仓，我们需要将元数据刷新自动化。如图所示，我们搭建了一个 pipeline ，当 Hive 中的表有更新时， Hive MetaStore 会记下一条 audit 日志。 Audit collector 将其发送到 Kafka ，然后被一个 Flink job 消费，触发 Impala 刷新缓存。

这条 pipeline 可以达到秒极延迟，但维护起来还是有点麻烦。幸运的是， Impala 在 3.2 版本引入了自动刷新元数据的功能（ IMPALA-7970 ），将来也会 merge 到 2.x 的版本中去。自动刷新元数据的功能还有许多细化的工作，具体见 IMPALA-7954 。这部分工作是 Cloudera 和 Hortonworks 合并之后才开始的，还处于起步阶段，大家可以关注一下。

3. Built-in的get_json_object函数  

get_json_object 是 Hive 中一个处理 JSON 字符串的函数，用于抽取 JSON 中的指定内容。 Impala 中并没有该函数的 native 实现，我们需要将 Hive 中实现该函数的 Java 类定义成 Impala 的 UDF 才能使用。在低版本的 Hive （ apache 版本小于 2.3 或 cdh 版本小于 5.12.0 ）中这个函数有内存泄漏的 bug （ HIVE-16196 ），而且 Impala 目前还没法追踪 各查询在JVM 里 所占用的内存，我们的 Impala 集群曾因此遭遇了 OOM 。为此我们实现了 native 的 get_json_object 函数，并贡献给了社区 (IMPALA-376) 。

总结

Impala 是 SQL-on-Hadoop 中一个高性能的 MPP 查询引擎。本文简要介绍了 Impala 的内部原理，以及 Hulu 在实际应用中对其做的一些优化和改进，包括增加对 ORC 文件格式的支持、外围的自动刷新元数据框架、支持 native 的 JSON 处理函数等。

大家在使用 Impala 中遇到的任何问题，欢迎加入Impala技术交流群 与我们探讨！同时也欢迎加入 Impala 社区的 SlackChannel（文末有链接） ！

如果二维码过期，请在微信公众号中回复”impala“获取最新二维码

参考文献与链接

• Cloudera Document, "Impala RuntimeFilter"”, www.cloudera.com/documentation/enterprise/latest/topics/impala_runtime_filtering.htm

• GoetzGraefe, " Encapsulation of Parallelism in the Volcano Query ProcessingSystem", http://daslab.seas.harvard.edu/reading-group/papers/encapsulation-volcano.pdf

• Impala Slack Channel, https://join.slack.com/t/apache-impala/shared_invite/enQtNTgzMzAyNzIyNTk0LTQwMzJjMDI0YzEwOWRmZDk2MzNlZTk5OWZkNTI4M2Y5MmU1MjQ1ZWIzYzQxMWQyMjUzNjNjNWU0NDQ1MTMyNWM

• "ORC Specification v1", https://orc.apache.org/specification/ORCv1/

• Mostafa Mokhtar, "Performance Optimizations inApache Impala", https://www.slideshare.net/cloudera/performance-of-apache-impala

作者简介：黄权隆，中国第一个Apache Impala PMC成员，毕业于北大计算机系网络所数据库实验室，目前就职于Hulu大数据基础架构团队，主要专注于OLAP引擎相关技术。