hadoop-impala十大优化之（7）—Impala查询运行时过滤最佳实践

1.1  Hadoop-impala

hadoop-impala十大优化之（7）—Impala查询运行时过滤最佳实践

1.1.1  运行时过滤

runtime_filter_mode=GLOBAL.

 运行时过滤是一种广泛的优化在CDH 5.7 / Impala 2.5及更高版本可用特性。只有当表中数据的一小部分是查询分区表或评价一个连接条件的需要， Impala 确定合适的条件在查询运行时，和广播信息的所有impalad节点，读表，可以避免不必要的I/O读取分区数据，并通过发送只有符合加入跨网络的键的行子集避免不必要的网络传输。

 此功能主要用于优化对大分区表（在名称动态分区修剪）和联接大表中的查询。本节中的信息，包括概念、内部，并为整个运行时故障排除信息过滤功能。对于分区表的特定调整步骤，参见动态分区修剪。

注：目前，此功能是随相应的设置设置为“medium”值（查询选项runtime_filter_mode=LOCAL）。这种保守的设置是为了避免回归现有的工作负载，由于潜在的更高的内存需求时，运行时过滤是完全启用。当你读到这一段，做一些性能和可扩展性的实验，通常你应该能够切换到“high”的runtime_filter_mode=GLOBAL。

1.1.2  运行时过滤的背景信息

要了解运行时过滤如何在一个详细的水平工作，您必须熟悉分布式数据库技术领域的一些术语：

 什么是计划分段。Impala的每一个查询分解成更小的工作单元，分布在集群。在可能的情况下，通过对同一主机上执行的计划分段进行读取、过滤和聚合的数据块。对于某些操作，如连接结合中间结果为最终结果，数据传输从一个节点在网络上的另一个。

 什么是扫描和哈希连接计划节点，以及它们在计算查询结果中的作用：

 在Impala的查询计划，扫描节点执行I/O从底层数据文件读取。虽然这是从传统的数据库来说一个昂贵的操作，Hadoop集群和Impala进行优化，在一个高度并行的方式做这种I/O。主要的潜在的成本节约来自使用柱状Parquet格式（如Impala可以避免不必要的列读取数据）和分区表（在Impala可以避免不必要的分区读取数据）。

 大多Impala联接使用哈希连接机制。（只是到最近的Impala开始使用嵌套循环连接技术，对非等值连接查询。某些种）在哈希联接，当评估加入条件从两个表，Impala和所有不同的列的值从表上加入一侧建有一内存中的哈希表。然后，为每一行从桌子上连接的另一边，Impala测试相关的列的值在这个哈希表或不。

 一个哈希连接节点构建这样一个内存哈希表，然后进行比较，以确定哪些行匹配相关的连接条件，并应包括在结果集（或至少发送到随后的中间阶段的查询处理）。因为一个哈希连接的输入可能会在网络上传输到另一个主机，它是特别重要的，从性能的角度来看，修剪提前任何数据是不相关的。

 更明显的值是在用作联接键的列中，在内存中的哈希表中较大的值，因此处理查询所需的内存越多。

 广播连接和洗牌连接之间的区别。（一个洗牌加入Hadoop的概念有时被称为Impala的分区加入。）在一个广播连接，从连接一侧的表（通常是较小的表）是发送所有的所有查询中所涉及的主机。然后每个主机可以比较它的数据的一部分，从其他（更大）表对全套可能的连接键。在一个混洗连接中，没有明显的“更小”的表，所以这两个表的内容都被划分了，并且数据的相应部分被传送给涉及查询的每个主机。在Impala的选择有关不同种类的连接处理报表看到查询提示。

 在构建阶段和探索阶段时，Impala过程联接查询的概念。生成阶段是包含连接键列的行，通常为较小的表，在网络上传输，并内置到一个或多个目标节点上的内存哈希表数据结构。探测阶段是在本地读取数据（通常来自较大的表），并将连接键列与内存哈希表中的值进行比较。相应的输入源（表，子查询，等等）这些阶段称为建设方和探头侧。

 如何设置查询选项：Impala在一个Impala交互式shell会话通过设置命令，一个JDBC或ODBC应用程序通过SET语句，或用于所有的impalad守护进程通过default_query_options配置设置。

1.1.3  运行时过滤组件

在计划分段之间传输的过滤器基本上是联接键列的值的列表。当这个列表值实时传送到扫描节点，Impala可以过滤掉非匹配值后立即阅读它们，而不是发送数据到另一台主机进行比较，对内存中的哈希表的主机上。该数据结构被实现为一个盛开的过滤器，它使用一个基于概率的算法，以确定所有可能的匹配值。（基于概率的方面意味着过滤器可能包括一些非匹配的值，但如果是的话，这不会导致在最终结果中的任何不准确。）

 有不同种类的过滤器来匹配不同类型的连接（分区和广播）。广播滤波器是一个完整的列表，可以立即通过扫描节点进行评估。分区的过滤器是一个相关值的部分列表（基于集群中的一个主机处理的数据）；所有分区的过滤器必须合并成一个（由协调器节点）在扫描节点可以使用的结果，以准确地过滤数据，因为它是从存储。

 广播过滤器也被归类为本地或全球。与本地广播过滤器，过滤器中的信息是由一个后续的查询片段，运行在同一主机上产生的过滤器。非本地广播筛选器必须在网络上传输到正在运行在不同主机上的查询片段。Impala指定了3个主机每个生产非本地广播的过滤器，以防止一个太长的单慢主机的可能性。根据不同的runtime_filter_mode查询选项的设置（本地或全球），Impala或用一个保守的优化策略筛选器只消耗在产生相同的主机，或者更激进的策略，过滤器可以通过网络传输。

1.1.4  运行时过滤的文件格式考虑

运行时过滤的文件格式考虑

 Parquet表最受益于运行时过滤优化。运行时过滤可以加快对分区或分区加入Parquet表的查询，和单表查询与Parquet表分区。请参阅使用Parquet格式的有关使用Parquet表Impala表信息。

 对于其他的文件格式（文本，Avro，RCFile，并通过），运行时过滤速度对分区表查询。因为分区表可以使用混合的格式，Impala在所有的情况下产生的过滤器，即使他们不是最终用于优化查询。

1.1.5  运行时过滤器的等待间隔

因为它需要时间来生产运行的过滤器，特别是分区的过滤器，必须由协调器节点的组合，有一个以上的时间间隔，它是更有效的扫描节点去构建自己的中间结果集，即使中间数据大于最优。如果只需要几秒钟产生的过滤器，它是值得的额外的时间，如果修剪不必要的数据，可以节省分钟，在整个查询时间。您可以指定最大等待使用runtime_filter_wait_time_ms查询选项，以毫秒为单位的时间。

默认情况下，每个扫描节点等待高达1秒（1000毫秒）的过滤器到达。如果所有的过滤器都没有到达指定的时间间隔内，扫描节点进行，使用任何过滤器的到达，以帮助避免读取不必要的数据。如果扫描节点开始读取数据后，如果一个过滤器到达，扫描节点将该过滤器应用到过滤器到达后读取的数据，但不适用于已经读过的数据。

如果群集相对繁忙，您的工作负载包含许多资源密集型或长时间运行的查询，可以考虑增加等待时间，以便复杂的查询不会错过优化的机会。如果群集是轻负载，您的工作负载包含了许多小的查询只花了几秒钟，考虑减少等待时间，以避免每个查询的1秒延迟。

1.1.6  运行时过滤的查询选项

有关查询选项，控制运行时过滤的查询选项的以下部分：

 第一个查询选项调整此功能的灵敏度-sensitivity。目前，默认情况下，它被设置为一个中等水平（本地-LOCAL）。验证后，更高的设置（全局-GLOBAL）不会导致任何与你现有的工作负载的回归，你通常应该使用全局设置为您的默认。

 runtime_filter_mode查询选项（CDH 5.7或更高）

 其他查询选项调整旋钮，通常只有调整后做性能测试，你可能想改变只有一个昂贵的查询时间：

 max_num_runtime_filters查询选项（CDH 5.7或更高）

 runtime_bloom_filter_size查询选项

 disable_row_runtime_filtering查询选项（CDH 5.7或更高）

1.1.7  运行时过滤和查询计划

运行时过滤和查询计划

 以相同的方式显示由解释语句所显示的查询计划包括每个计划片段使用的谓词的信息，它还包括说明是否一个计划片段产生或消耗一个运行时过滤器的注释。计划的片段产生滤波器包含注释如运行过滤器：filter_id <- table.column，而计划的片段，消耗了滤波器包含注释如运行过滤器：filter_id -> table.column。

 下面的示例显示了一个查询，使用一个单一的运行时过滤（标记rf00）修剪是一个阶段的查询扫描分区，以查询结果的评价：

createtable yy (s string) partitioned by (year int) stored as parquet;
insert intoyy partition (year) values ('1999', 1999), ('2000', 2000),
  ('2001', 2001), ('2010',2010);
computestats yy;
 
createtable yy2 (s string) partitioned by (year int) stored as parquet;
insert intoyy2 partition (year) values ('1999', 1999), ('2000', 2000),
  ('2001', 2001);
computestats yy2;
 
-- Thequery reads an unknown number of partitions, whose key values are only
-- known atrun time. The 'runtime filters' lines show how the information about
-- thepartitions is calculated in query fragment 02, and then used in query
-- fragment00 to decide which partitions to skip.
explainselect s from yy2 where year in (select year from yy where year between 2000and 2005);
+----------------------------------------------------------+
| ExplainString                                          |
+----------------------------------------------------------+
| EstimatedPer-Host Requirements: Memory=16.00MB VCores=2 |
|                                                          |
|04:EXCHANGE [UNPARTITIONED]                              |
| |                                                       |
| 02:HASHJOIN [LEFT SEMI JOIN, BROADCAST]                |
| |  hash predicates: year = year                          |
| |  runtime filters: RF000 <- year                       |
| |                                                       |
||--03:EXCHANGE [BROADCAST]                               |
| |  |                                                     |
| |  01:SCAN HDFS [dpp.yy]                                 |
| |     partitions=2/4 files=2 size=468B                   |
| |                                                       |
| 00:SCANHDFS [dpp.yy2]                                   |
|    partitions=2/3 files=2 size=468B                      |
|    runtime filters: RF000 -> year                       |
+--------------------------------------------

查询文件概要（在 impala-shel命令显示概要）包含解释计划和查询的内部运作，更详细的信息。该配置文件输出包括一个标记的“过滤器路由表”，关于每个过滤器的信息的基础上它的身份。

1.1.8  从运行时过滤中获益的查询示例

在这个例子中，Impala通常会做额外的工作来解释列C1，C2，C3，和我在huge_t1每一行，在检查ID值在内存中的哈希表构造的所有tiny_t2.id值。通过产生一个过滤器包含所有tiny_t2.id值甚至在查询开始扫描huge_t1表，Impala可以跳过不必要的工作来解析列信息尽快确定一个ID值不匹配任何值从其他表。

 实例表明，计算统计报表的表（尽管这是一次操作后将数据加载到表）因为Impala依靠最新的统计数据来确定哪一个更具有独特的ID值比其他。这些信息让Impala做出有效的决定哪些表使用构建在内存中的哈希表，并从磁盘中读取并比较对哈希表中的表项。

setruntime_filter_mode=global;
COMPUTESTATS huge_t1;
COMPUTESTATS tiny_t2;
SELECT c1,c2, c3 FROM huge_t1 JOIN tiny_t2 WHERE huge_t1.id = tiny_t2.id;

在这个例子中，T1是一个表分区的年。在T2的子查询产生的多个值，并将这些值作为过滤器，阅读从T1计划片段。任何不匹配的分区在T1被跳过。

setruntime_filter_mode=global;
select c1from t1 where year in (select distinct year from t2);

现在，其中包含一个不适用于分区键列的附加测试。在不是一个分区键的列上的一个过滤器被称为每行筛选器。因为每行筛选器仅适用于Parquet，Parquet表T1必须。

 在两个过滤器被传送到扫描节点读取T1的子查询结果。一年的过滤器有助于查询消除基于非匹配年的整个分区。在C2的过滤让Impala丢弃不匹配的C2值后立即阅读他们的行。没有运行时过滤，Impala必须保持非匹配值在内存中，C1 C2和C3，组装，成排的中间结果集，并发送所有的中间排到协调器节点，在那里他们将只在查询末尾淘汰。

setruntime_filter_mode=global;
select c1,c2, c3 from t1
  where year in (select distinct year from t2)
    and c2 in (select other_column from t3);

这个例子涉及一个广播连接。事实上，在条款会返回一小部分匹配的行（因为没有在tiny_t2很多行）意味着相应的滤波器是非常有选择性的。因此，运行时过滤可能是有效的优化这个查询。

setruntime_filter_mode=global;
select c1from huge_t1 join [broadcast] tiny_t2
  on huge_t1.id = tiny_t2.id
  where huge_t1.year in (select distinct yearfrom tiny_t2)
    and c2 in (select other_column from t3);

这个例子涉及shuffle 或 partitioned连接。假设在huge_t1最行有huge_t2相应的行。事实上，在子句可以返回一个大量的匹配行意味着相应的过滤器将不会是非常有选择性的。因此，运行时过滤可能会在优化此查询时效果较差。

setruntime_filter_mode=global;
select c1from huge_t1 join [shuffle] huge_t2
  on huge_t1.id = huge_t2.id
  where huge_t1.year in (select distinct yearfrom huge_t2)
    and c2 in (select other_column from t3);

 

1.1.9  调整和排除使用运行时过滤的查询

这些调整和故障排除程序适用于资源密集型的查询，运行足够长的时间，以及足够频繁的，你可以特别注意优化它们。

 使用解释语句并检查运行时筛选器：确定运行时筛选器是否被应用到您所期望的谓词和连接子句中。例如，运行时过滤不适用于查询使用嵌套循环连接由于非等值连接运营商的机制。

 确保所有表中涉及的所有表都是最新的。在将数据加载到非分区表后使用计算统计报表，并在将新分区添加到分区表后计算增量属性。

 如果加入涉及大表查询使用独特的柱作为连接键，例如加入一个主键列和外键列，生产和发射滤波器的开销可能会超过履行利益因为没有太多的数据可以被修剪在查询的早期阶段。对于这样的疑问，考虑设置查询选项runtime_filter_mode =OFF。

1.1.10     运行时过滤的限制和限制

运行时过滤功能是最有效的Parquet 文件格式。对于其他文件格式，只适用于分区表。请参见运行时过滤的文件格式考虑。

 当溢出到磁盘机制被激活时，在一个特定的主机在查询过程中，主机不产生任何过滤器，而处理该查询。此限制不影响结果的正确性，它只减少了可以应用于查询的优化量。