Apache Impala的新多线程模型

简介

今天，我们将介绍一系列新的博客文章，其中将介绍Apache Impala的最新增强功能，许多是性能改进，例如下面所述的功能，可以更高效地利用所有CPU内核，将性能提高2倍至7倍。此外还有大量的工作，确保Impala在存储计算分离的场景中能最优地运行，比如数据存储在对象存储或远程HDFS的场景。鉴于越来越多的用户正在运行容器化的Impala集群，例如Cloudera Data Warehouse（CDW）服务中提供的集群，这一点尤其重要。

图1.使用Apache Impala的新多线程模型的性能提升（20个Executor，mt_dop = 12）

新的多线程模型的目的

在第一篇文章中，我们将重点介绍在查询执行方面最近完成的工作，就是扩展查询执行里的多线程模型。首先交代一些相关的上下文，Impala的设计理念的两个主要原则是：

• 并行性––对于查询执行的每个部分，在尽可能多的资源上并行运行它
• 开放的文件格式––提供对Apache Parquet和ORC等开放的源文件格式的原生查询，以防止锁定并鼓励互操作性

从第一天起，Impala就能够分解查询并在多个节点上运行它––真正的大规模并行处理（MPP）引擎。这是水平扩展（scale out）的极致。几年前，我们添加了多线程扫描，从而使每个节点上的多个线程可以同时扫描所需的数据。通过增加在节点内进行垂直扩展（scale up）的能力，进一步提高了并行度，尽管仅限于有限的几个操作。到今天为止，我们现在能够对Join和Aggregation进行多线程处理，从而使我们可以并行执行查询执行的每个部分––通过水平扩展和垂直扩展。通过此项改进，如果输入数据足够大（即足够的行或不同的值），则查询中的关键操作可以在节点内垂直扩展。

这是跟开放文件格式相关的一点。目前没有其他的云数据仓库引擎，能对存储在对象存储里的开放文件格式提供原生查询，在Join和Aggregation上达到能跟Impala相抗衡的性能。这些是分析查询中的常见瓶颈，并且众所周知难以优化。

大多数查询引擎通过利用查询层和存储层之间的紧密耦合，在Join和Aggregation级别上实现了性能改进。它们的“秘方”通常会导致专有文件格式和/或需要本地存储。因此是以灵活性、互操作性和成本为代价来提高性能。Impala是第一个将此类SQL优化与云存储上下文中的开放文件格式有效地结合在一起的SQL引擎。

新的多线程模型如何工作

如果您觉得太长读不下去了，可以简单了解以下要点：

• 就像我们在节点间做的并行化一样，我们在节点内运行多个fragment实例来提高并行度。
• 每个实例都是单线程执行，并通过Exchange运算符将数据发送到其他实例–节点间通信使用相同的机制。
• 查询或Session级别的并行度（DOP）由一个mt_dop（MultiThreading Degree of Parallelism）参数控制。

如果您需要更多详细信息，请继续阅读...

计划生成

Impala的现有计划生成过程分两个阶段进行：首先是从分析输出中生成单节点计划；然后将其转换为分布式计划。使用此新的多线程模型，添加了第三阶段。分布式计划（它是计划碎片的树）被转换为并行计划（它是分布式计划树）。分布式计划由Join的Build算子连接（如为Hash Join构建内存哈希表），基于这些Build算子，父计划中的Join操作得以在内存中流式执行。

下图显示了使用TPC-H查询11的示例。第一个是分布式计划，其中每个框都是片段（fragment）。

第二个是相应的并行计划，其中：

• 计划树的根计划（计划1）以蓝色显示
• 它的直属子组Group 1就有额外的并行计划阶段，以绿色显示
• Build操作的输出ID以红色显示
• Build [0] 计划的直属子节点组成了Group 2，显示为橙色

调度器

Impala调度器的输入是一个并行计划，该计划已分为多个片段（fragment），其输出是要运行这些片段的实例数以及在哪些节点上运行。该过程总结如下。

1. 根据数据本地性（locality）和调度器的负载均衡启发式规则，将扫描范围（scan range）分配给各个节点。
2. 在这些节点上创建片段实例（fragment instance），并在实例之间划分扫描范围。mt_dop限制了每个节点将创建的最大实例数。

从上面的示例中，如果partsupp表被划分为32个均匀大小的远程scan range并在4个节点上以mt_dop = 4运行，则可能为每个节点分配8个scan range（取决于数据位置），每个节点最多可运行4个F6实例。然后，每个F6实例由不同的线程执行，在每个节点上并行使用4个CPU。在下面的图中，每个片段实例均由白色正方形表示，该白色正方形说明了此扫描操作的水平扩展和垂直扩展。

配置项

使新的多线程模型的配置项保持简单是一个明智的决定，这就是只设计一个mt_dop选项的原因。它确定查询的最大并行度，如果查询较小，Impala将自动降低并行度。可以在服务器、资源池、会话（session）和查询级别设置此选项。除了为用户提供调整节点内并行度级别的旋钮外，这还具有重要的副作用，可以使调度变得更容易，查询延迟也更可预测。

准入控制（Admission Control）

引入mt_dop选项意味着Impala查询可能具有非常不同的CPU需求。因此，我们必须将此因素纳入准入控制决策中，以避免资源浪费和过饱和。例如，更多的并发查询可以以低DOP运行，而不是高DOP，因为内核不会超额认购。

Impala具有“准入控制插槽”的概念-Impala daemon所允许的并行度。默认为处理器数量，可以使用–admission_control_slots来配置。准入控制插槽模型为准入控制和多线程执行提供了最佳道路。

在executor上运行的查询将消耗有效dop（degree of parallelism）个插槽，该有效dop计算为在执行程序上运行的片段的最大实例数。例如，如果有2个F0实例和4个F1实例，则有效dop为4。

在某些情况下，需要加大Impala负载以最大化CPU利用率，或相反地，通过降低负载来实现可预测性。可以通过将–admission_control_slots更改为大于或等于系统上的内核数来进行管理。

对资源消耗的影响

我们已设法最小化或消除与这种新的多线程模型在CPU、内存和网络方面的额外开销。以下是一些值得注意的要点：

• I/O像过去一样继续受益于多线程，并且此新模型没有引入额外的开销。
• 高基数grouping aggregations可能会对总网络流量产生某些影响。
• 由于Hash Join、Hash Aggregation、sort、analytic和runtime filter都需要预留内存，我们将看到每个额外的片段实例的最小内存预留将有一定的增加–通常每个实例1MB到40MB。如果内存限制设置得很低或查询计划非常复杂，则用户可以降低mt_dop来解决此问题。
• 由于线程数量的增加，查询的峰值内存需求可能会有所增加，但是我们努力通过减少每个线程的内存需求来避免大多数查询的显着增加。通常可以减少每个查询的总内存消耗，因为查询可以更快完成并释放资源。

查询执行影响示例

在本节中，我们将看一些新的多线程模型对执行过程各个步骤产生影响的示例。这给出了实现细节的思想，以及为减少使用多线程模型所需的CPU和内存开销而进行的工作。

Grouping Aggregation

Grouping Aggregation通过以下方式并行化：a）在每个片段实例中复制预聚合操作（pre-aggregation），以及b）在所有片段实例中将输入分区到合并聚合操作（merge-aggregation）。

Broadcast Hash Join

对于Broadcast Hash Join，并行化基于将Build阶段和探查（Probe）阶段分离为不同的计划。前者构建Hash Join和Nested Loop Join所需的数据结构，后者探查它们并生成结果行。

Exchange

新的并行化模型会导致每个Exchange操作的相关片段实例数量大大增加。例如，以前用于Partitioned Join的Hash Exchange将具有最多是节点数目个实例的数据流；现在它具有来自节点数目* dop个实例的数据流。为了解决这一点，KRPC将所有这些逻辑流（logical stream）多路复用到每对节点之间的单个点对点连接中，从而避免了许多潜在的可伸缩性问题。

分析函数（Analytic Function）

分析函数由PARTITION BY子句确定并行性。输入行通过hash分区分发到各个实例，然后独立计算每个分区。仅当分区数多于节点数时，才可能实现多线程加速。在这种情况下，由于每个分区的计算都是独立的，因此可以实现线性加速。

运行时代码生成（Runtime Code Generation）

Impala中的运行时代码生成原本是为每个片段实例执行的。这种方法仍然可以与新的多线程模型一起使用，但是会很浪费–它冗余地为每个片段实例生成代码dop次。相反，我们重构了代码生成，以便对impala daemon上同一片段的所有实例只执行一次。使运行时代码生成的影响最小化的其他改进包括最近实现的异步代码生成改进（IMPALA-5444），以及当前正在进行的有关代码生成缓存的工作。

Runtime Filter

Runtime Filter当前由相关片段实例生成，然后发送到Coordinator，然后由Coordinator将它们独立地发送到所有相关节点上的每个片段。由于使用了多线程，用于Partitioned Join的Filter数量将增加dop倍，可能导致Coordinator成为瓶颈。为了防止这种情况，我们在发送给Coordinator之前在每个节点上本地聚合Partitioned Join的Filter，以使在Coordinator上完成的工作总量不取决于并行度。

多线程效果的样例展示

为了深入研究多线程如何影响您的查询性能，我们以下面的TPC-DS基准测试中的查询84为例。该查询涉及将两个大型事实表与四个维度表连接在一起。下面的查询计划图表明颜色鲜艳的操作占用了查询执行时间的最大部分。在这种情况下，大部分时间专用于执行Join操作，这意味着查询可能比I / O更容易受到CPU的瓶颈，这使其成为多线程执行的理想选择。

为了查看此查询在更多CPU内核上的扩展效果如何，我们使用等于1的dop运行了该查询，然后逐步提高了配置的并行度。该测试是在CDW服务上完成的，使用具有16个vCPU（r5d.4xlarge）的实例查询存储为S3中Apache Parquet文件的10TB TPC-DS数据集。

使用从dop=1的运行时作为基线，我们可以看到性能几乎呈线性增长，直到4级。但是，我们在8处略有下降，而并行度为12，我们没有做优于7.5倍的性能提升。许多因素可能是收益递减的因素，包括数据偏斜，16个虚拟CPU与8个物理CPU或后面各节中讨论的其他实现效率低下的问题。我们认为这是一个不错的开始，并且我们看到了许多继续改善Impala并行可伸缩性的机会。我们知道选择正确的并行度可能会对最终用户或Impala管理员造成麻烦，因此我们计划在将来的版本中努力在查询执行期间自动确定最佳值。

性能分析

在内部TPC-DS基准测试中，分析了这种新的多线程模型（比较mt_dop = 1与mt_dop = 12）所带来的性能改进，我们看到：

• 所有99个查询的总运行时间提升了2.43倍（从7,549秒到3,104秒）
• 所有99个查询单独运行时间平均提升1.66倍（从16.38到9.84）。译者注：16.38可能是1.638的笔误。

下表显示了运行时间有2倍以上提升的36个查询，以举例说明我们在哪些方面获得了最大的收益。

值得一提的有关此测试的一些重要说明：

• 对于Aggregation或Hash Join在数据偏斜键上的查询，多线程的好处并不那么明显，因为只有极少数的节点能够从更高的CPU利用率中受益（即具有数据偏斜键值的节点）
• 对于短查询（定义为不使用新的多线程模型就已经能运行时间少于5秒的查询），运行时的改进不太明显。这主要是因为在执行期间的其他步骤（例如，生成执行计划，profile的聚合）中花费的时间往往在短时间查询中占主导地位。我们也在努力提高其他方面的效率，以IMPALA-9378为例。
• 相反，对于更长的查询，运行时的改进更为明显。
• 请记住，Impala以前以多线程方式运行了查询执行的某些部分，例如数据扫描，文件I/O，Join的Build端以及某些流水线查询片段。因此，某些查询，尤其是那些瓶颈在扫描阶段的查询，已经高度并行化，并且在新的多线程模型中的改进空间较小。
• 以及是的，最近Apache Impala添加了对SQL功能的支持，使其能够运行所有99个TPC-DS查询。

APACHE一些很棒的SQL新功能在APACHE IMPALA 4.0之前的MASTER中引入– INTERSECT、EXCEPT、ROLLUP、CUBE、GROUPING SETS，以及更多的子查询支持。

-APACHE IMPALA（@APACHEIMPALA）2020年7月31日

讨论此功能的影响

这种新的多线程模型将在以下情况下提供最大的好处：

• 并行运行有限数量查询的工作负载–因为查询执行的大多数方面以前都是单线程的，所以在低并发性下很难实现较高的CPU利用率。一旦大多数查询执行路径都是多线程的，那么相同的低并发工作负载将能够利用更多的CPU内核。
• 高CPU集群–由于计算节点上的CPU内核（例如48个内核）非常密集，即使具有更高的并发级别，也很难实现较高的CPU利用率。这样，无论哪种情况，Impala都能充分利用大量内核。
• 瓶颈在计算上的工作负载（Compute-bound workloads）–对于瓶颈在计算上的查询，除非CPU使用率已经最大化，否则使用这种新的多线程模型它们将运行得更快。

另一方面，在扫描密集型查询中，可以期望看到的改进较少，例如使用LIKE运算符或执行regexp_extract搜索字符串列。因为扫描已经是多线程的，所以没有更多的CPU使用率收益。

在云环境中运行Impala时，此优化变得尤为重要，因为在该环境中，计算集群可以自动启动和停止，或者扩展和收缩节点。当您的工作负载运行得更快时...

• 自动挂起的DW越早关闭，您为此支付的费用就越少
• 自动缩放DW越早缩小，您支付的费用就越少，因为它使用的计算资源更少

此外，如果您可以针对给定的工作负载使用更多数量的可用核心，则可能可以使用更少数量的计算节点。这降低了总成本。

最后，鉴于在云中轻松配置Impala集群（这对于工作负载隔离和自助服务敏捷性很重要），通常会看到大量专用于单个工作负载的集群，这些集群可能无法在服务器的CPU级别上得到充分利用。这种优化可确保即使在这种情况下，工作负载也可以实现较高的利用率。

总结

回顾一下……Impala现在可以在分析型查询中对某些最重量级的操作进行多线程处理，即Join和Aggregation。我们做到了这一点，并且没有牺牲使用开放文件格式的能力，也不需要本地化存储。这将带来更高的CPU利用率、更快的查询时间和更低的云成本。

您可以通过Cloudera的CDP试用版经验自己尝试一下。

转到Cloudera工程博客，详细了解Impala的性能和架构。

致谢

特别感谢构建此新多线程模型的许多人。Tim Armstrong，Joe McDonnell，Bikramjeet Vig，David Rorke和Kurt Deschler参与了该项目的最新阶段。为可伸缩性和多线程做基础工作的其他Impala提交者和贡献者包括Marcel Kornacker和Alex Behm。

原文作者：Tim Armstrong& David Rorke & Shant Hovsepian & Justin Hayes

原文链接：https://blog.cloudera.com/new-multithreading-model-for-apache-impala/