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

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

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

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

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

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

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

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

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

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

引入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、内存和网络方面的额外开销。以下是一些值得注意的要点：

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

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

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

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

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

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

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）所带来的性能改进，我们看到：

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

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

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

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

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

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

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

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

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

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