Apache Doris Join 优化原理详解

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

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

　　背景 & 目标
　　掌握 Apache Doris Join 优化手段及其实现原理
　　为代码阅读提供理论基础

　　Doris 数据划分
　　不同的 Join 方式非常依赖于对 Doris 中数据划分方式的透彻理解。因此先在这里列举出必要的基础知识。

　　首先，在 Doris 中数据都以表（Table）的形式进行逻辑上的描述。

　　在 Doris 的存储引擎中，用户数据被水平划分为若干个数据分片（Tablet，也称作数据分桶 Bucket）。每个 Tablet 包含若干数据行。各个 Tablet 之间的数据没有交集，并且在物理上是独立存储的。

　　一个 Tablet 只属于一个数据分区（Partition）。而一个 Partition 包含若干个 Tablet。因为 Tablet 在物理上是独立存储的，所以可以视为 Partition 在物理上也是独立的。Tablet 是数据移动、复制等操作的最小物理存储单元。

　　若干个 Partition 组成一个 Table。Partition 可以视为是逻辑上最小的管理单元。数据的导入与删除，仅能针对一个 Partition 进行。

　　Doris 支持两层的数据划分。第一层是 Partition，支持 Range 和 List 的划分方式。第二层是 Bucket（Tablet），仅支持 Hash 的划分方式。也可以仅使用一层分区。使用一层分区时，只支持 Bucket 划分。

　　下图说明 Table、Partition、Bucket（Tablet）的关系：

　　Table 按照 Range 的方式按照 date 字段进行分区，得到了 N 个 Partition
　　每个 Partition 通过相同的 Hash 方式将其中的数据划分为 M 个 Bucket（Tablet）
　　从逻辑上来说，Bucket 1 可以包含 N 个 Partition 中划分得到的数据，比如下图中的 Tablet 11、Tablet 21、Tablet N1

　　特别注意：

　　Doris 中的 Partition 和 Bucket 定义可能和某些其它数据库系统的定义有一些差异，下面配以一个具体的建表语句为例来说明：

　　1
　　2
　　3
　　4
　　5
　　6
　　7
　　8
　　9
　　10
　　11
　　12
　　13
　　14
　　15
　　16
　　17
　　18
　　19
　　20
　　21
　　22
　　23
　　24
　　25
　　26
　　CREATE TABLE IF NOT EXISTS example_db.expamle_range_tbl
　　(
　　    `user_id` LARGEINT NOT NULL COMMENT "用户id",
　　    `date` DATE NOT NULL COMMENT "数据灌入日期时间",
　　    `timestamp` DATETIME NOT NULL COMMENT "数据灌入的时间戳",
　　    `city` VARCHAR(20) COMMENT "用户所在城市",
　　    `age` SMALLINT COMMENT "用户年龄",
　　    `sex` TINYINT COMMENT "用户性别",
　　    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "用户最后一次访问时间",
　　    `cost` BIGINT SUM DEFAULT "0" COMMENT "用户总消费",
　　    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "用户最大停留时间",
　　    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "用户最小停留时间"
　　)
　　ENGINE=OLAP
　　AGGREGATE KEY(`user_id`, `date`, `timestamp`, `city`, `age`, `sex`)
　　PARTITION BY RANGE(`date`)
　　(
　　    PARTITION `p201701` VALUES LESS THAN ("2017-02-01"),
　　    PARTITION `p201702` VALUES LESS THAN ("2017-03-01"),
　　    PARTITION `p201703` VALUES LESS THAN ("2017-04-01")
　　)
　　DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
　　PROPERTIES
　　(
　　    "replication_num" = "3"
　　);
　　绿色高亮：Partition，此例中使用一个 date 字段进行分区

　　蓝色高亮：Bucket，此例中使用 user_id 字段为作为分布列

　　Partition
　　Partition 列可以指定一列或多列，分区列必须为 KEY 列
　　分区数量理论上没有上限
　　当不使用 Partition 建表时，系统会自动生成一个和表名同名的，全值范围的 Partition。该 Partition 对用户不可见，并且不可删改
　　创建分区时不可添加范围重叠的分区

　　有两种分区方式：

　　分区方式一般用法
　　Range 通常按时间分区，以方便地管理新旧数据
　　List 支持的类型更丰富，分区值为枚举值。只有当数据为目标分区枚举值其中之一时，才可以命中分区

　　Bucket
　　如果使用了 Partition，则 DISTRIBUTED 语句描述的是数据在各个分区内的划分规则。如果不使用 Partition，则描述的是对整个表的数据划分规则
　　分桶列的选择，是在查询吞吐和查询并发之间的一种权衡：
　　如果选择多个分桶列，则数据分布更均匀。如果一个查询条件不包含所有分桶列的等值条件（意味着无法做桶裁剪以减少数据查询范围），那么该查询会触发所有分桶同时扫描，这样查询的吞吐会增加，单个查询的延迟随之降低。这个方式适合大吞吐低并发的查询场景
　　如果仅选择一个或少数分桶列，则对应的点查询可以仅触发一个分桶扫描（意味着可以做桶裁剪以减少数据查询范围）。此时，当多个点查询并发时，这些查询有较大的概率分别触发不同的分桶扫描，各个查询之间的 IO 影响较小，尤其当不同桶分布在不同磁盘上时），所以这种方式适合高并发的点查询场景
　　分桶的数量理论上没有上限

　　Join 方式

　　总览
　　作为分布式的 MPP 数据库，在 Join 的过程中是需要进行数据的 Shuffle。数据需要进行拆分调度，才能保证最终的 Join 结果是正确的。举个简单的例子，假设关系 S 和 R 进行Join，N 表示参与 Join 计算的节点的数量；T 则表示关系的 Tuple 数目。

　　目前 Doris 支持的 Join 方式有以上 4 种，这 4 种方式灵活度和适用性是从高到低的，对数据分布的要求越来越严，但 Join 计算的性能则通过降低网络开销而越来越好。

　　Join 方式的选择是 FE 生成分布式计划阶段会考虑的事项之一。在 FE 进行分布式计划时，优先选择的顺序为（总是会优先选择预期性能最好的）：Colocate Join -> Bucket Shuffle Join -> Broadcast Join -> Shuffle Join。

　　Colocate 以及 Bucket Shuffle 是可遇不可求的。当无法使用它们时，Doris会自动尝试进行 Broadcast Join，如果预估小表过大则会自动切换至 Shuffle Join。

　　但是用户可以通过显式 Hint 来强制使用期望的 Join 类型，比如：

　　1
　　select * from test join [shuffle] baseall on test.k1 = baseall.k1;

　　Broadcast / Shuffle Join
　　原理比较简单，这里不展开。

　　Bucket Shuffle Join
　　当 Join 条件命中了左表的数据分布列时，Broadcast 以及 Shuffle Join 会有非必要的网络传输开销。而 Bucket Shuffle Join 旨在解决这类问题，通过对左表实现本地性计算优化，来减少左表数据在节点间的传输耗时，从而加速查询。

　　以上的例子中，Join 的等值表达式命中了表 A（左表）的数据分布列。Bucket Shuffle Join 会根据表 A 的数据分布信息，将表 B（右表）的数据发送到对应表 A 的数据计算节点。

　　定性分析上：

　　降低了网络与内存开销（相比 Broadcast 以及 Shuffle Join 都不会更差），使一类 Join 查询有更好的性能。尤其是当 FE 能够执行左表的分区裁剪与桶裁剪时
　　与 Colocate Join 不同，它对于表的数据分布方式没有侵入性，对于用户来说是透明的。对于表的数据分布没有强制性的要求（体现在建表语句中不需要显式地设置 colocate_with 属性），不容易导致数据倾斜的问题
　　可以为 Join Reorder 提供更多可能的优化空间

　　Plan Rule
　　Bucket Shuffle Join 只生效于 Join 条件为等值的场景，原因与 Colocate Join 类似，它们都依赖 Hash 来计算确定的数据分布
　　在等值 Join 条件之中包含两张表的分桶列，当左表的分桶列为等值的 Join 条件时，它有很大概率会被规划为 Bucket Shuffle Join
　　由于不同的数据类型的 Hash 值计算结果不同，所以 Bucket Shuffle Join 要求左表的分桶列的类型与右表等值 Join 列的类型需要保持一致，否则无法进行对应的规划
　　Bucket Shuffle Join 只作用于 Doris 原生的 OLAP 表，对于 ODBC，MySQL，ES 等外表，当其作为左表时是无法规划生效的
　　对于分区表，由于每一个分区的数据分布规则可能不同，所以 Bucket Shuffle Join 只能保证左表为单分区时生效。所以在 SQL 执行之中，需要尽量使用 where 条件使分区裁剪的策略能够生效

（编辑：甘南站长网）

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

1/3

尾页