Apache Doris Join 优化原理详解

发布时间：2023-02-17 13:10:50 所属栏目：Apache 来源：互联网

导读：背景目标掌握 Apache Doris Join 优化手段及其实现原理为代码阅读提供理论基础 Doris 数据划分不同的 Join 方式非常依赖于对 Doris 中数据划分方式的透彻理解。因此先在这里列举出必要的基础知识。首先，在 Doris 中数据都以表（Table）的形式进行逻辑

　　假如左表为 Colocate 的表，那么它每个分区的数据分布规则是确定的，Bucket Shuffle Join 能在Colocate 表上表现更好

　　Colocate Join
　　可以理解为在数据分布满足一定条件的前提下，减少一切不必要的网络传输开销，实现完全的计算本地化来加速查询。同时因为没有网络传输开销，BE 节点可以拥有更高的并发度，从而进一步提升 Join 性能。

　　要理解这个算法，需要先了解两个术语：

　　Colocation Group（CG）：一个 CG 中会包含一张及以上的 Table。在同一个 Group 内的 Table 有着相同的 Colocation Group Schema，并且有着相同的数据分片分布
　　Colocation Group Schema（CGS）：用于描述一个 CG 中的 Table，和 Colocation 相关的通用 Schema 信息。包括分桶列类型，分桶数以及副本数等
　　和 Buckets Sequence 这一概念：

　　一个表的数据，最终会根据分桶列值 Hash、对桶数取模后落在某一个分桶内。假设一个 Table 的分桶数为 8，则共有 [0, 1, 2, 3, 4, 5, 6, 7] 8 个分桶（Bucket），我们称这样一个序列为一个 BucketsSequence。每个 Bucket 内会有一个或多个数据分片（Tablet）。当表为单分区表时，一个 Bucket 内仅有一个 Tablet。如果是多分区表，则会有多个（因为多个 Partition 中的不同 Tablet 会被划分到相同的 Bucket）。

　　Colocation Join 功能，是将一组拥有相同 CGS 的 Table 组成一个 CG。并保证这些 Table 对应的数据分片会落在同一个 BE 节点上。使得当 CG 内的表进行分桶列上的 Join 操作时，可以通过直接进行本地数据 Join，减少数据在节点间的传输耗时。

　　

　　因此关键问题就转变为了「如何保证这些 Table 对应的数据分片会落在同一个 BE 节点上？」

　　通过同一 CG 内的 Table 必须保证以下属性相同实现：

　　分桶列和分桶数
　　分桶列，即在建表语句中 DISTRIBUTED BY HASH(col1, col2, ...) 中指定的列。分桶列决定了一张表的数据通过哪些列的值进行 Hash 划分到不同的 Tablet 中。同一 CG 内的 Table 必须保证分桶列的类型和数量完全一致，并且桶数一致，才能保证多张表的数据分片能够一一对应的进行分布控制。

　　副本数
　　同一个 CG 内所有表的所有分区（Partition）的副本数必须一致。如果不一致，可能出现某一个 Tablet 的某一个副本，在同一个 BE 上没有其他的表分片的副本对应。不过，同一个 CG 内的表，分区的个数、范围以及分区列的类型不要求一致。

　　

　　在固定了分桶列和分桶数后，同一个 CG 内的表会拥有相同的 BucketsSequence。而副本数决定了每个分桶内的 Tablet 的多个副本，存放在哪些 BE 上。假设 BucketsSequence 为 [0, 1, 2, 3, 4, 5, 6, 7]，BE 节点有 [A, B, C, D] 4个。则一个可能的数据分布如下：

　　1
　　2
　　3
　　4
　　5
　　6
　　7
　　8
　　9
　　+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
　　| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |
　　+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
　　| A | | B | | C | | D | | A | | B | | C | | D |
　　|   | |   | |   | |   | |   | |   | |   | |   |
　　| B | | C | | D | | A | | B | | C | | D | | A |
　　|   | |   | |   | |   | |   | |   | |   | |   |
　　| C | | D | | A | | B | | C | | D | | A | | B |
　　+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
　　CG 内所有表的数据都会按照上面的规则进行统一分布，这样就保证了，分桶列值相同的数据都在同一个 BE 节点上，可以进行本地数据 Join。其核心思想是「两次映射」，保证相同的 Distributed Key 的数据会被映射到相同的 Bucket Sequence，再保证 Bucket Sequence 对应的 Bucket 映射到相同的 BE 节点：

　　通过查询计划可以检查一个查询是否使用了 Colocate Join，同时计划中的 Exchange Node 也被去掉了，会将 ScanNode 直接设置为 Hash Join Node 的孩子节点。

　　DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);
　　-- 在 Hash Join 节点会显示：
　　-- colocate: true/false
　　Colocate Join 十分适合几张表按照相同字段分桶，并高频根据固定的字段 Join 的场景。这样可以将数据预先存储到相同的分桶中，实现本地计算。

　　Runtime Filter 优化
　　Doris 在进行 Hash Join 计算时会在右表构建一个 Hash Table，左表流式地通过右表的 Hash Table 从而得出 Join 结果。而 Runtime Filter 就是充分利用了右表的 Hash Table 构建阶段去做一些额外的事情。

　　在右表生成 Hash Table 的时，同时生成一个基于 Hash Table 数据的一个过滤条件，然后下推到左表的数据扫描节点。通过这样的方式，Doris 可以在运行时进行数据过滤。

　　假如左表是一张大表，右表是一张小表，那么利用下推到左表的过滤条件就可以把绝大多数 Join 层要过滤的数据在数据读取时就提前过滤（如果能够下推到引擎层，还能够利用 Doris 针对 Key 列过滤的延迟物化），从而大幅度地提升 Join 查询的性能。

　　Runtime Filter 在查询规划时生成，在 HashJoinNode 中构建，在 ScanNode 中应用。比如 T1（行数 10w）和 T2（行数 2k）的 Join 操作：

　　
　　|          >      HashJoinNode     <
　　|         |                         |
　　|         | 100000                  | 2000
　　|         |                         |
　　|   OlapScanNode              OlapScanNode
　　|         ^                         ^
　　|         | 100000                  | 2000
　　|        T1                        T2
　　|
　　显而易见对 T2 扫描数据要远远快于 T1，如果我们主动等待一段时间再扫描 T1，等 T2 将扫描的数据记录交给 HashJoinNode 后，HashJoinNode 根据 T2 的数据计算出一个过滤条件，比如 T2 数据的最大和最小值，或者构建一个 Bloom Filter，接着将这个过滤条件发给等待扫描 T1 的 ScanNode，后者应用这个过滤条件，将过滤后的数据交给 HashJoinNode，从而减少 probe hash table 的次数和网络开销，这个过滤条件就是 Runtime Filter，效果如下:

　
　　|          >      HashJoinNode     <
　　|         |                         |
　　|         | 6000                    | 2000
　　|         |                         |
　　|   OlapScanNode              OlapScanNode
　　|         ^                         ^
　　|         | 100000                  | 2000

（编辑：甘南站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!

2/3

首页

尾页