法国化妆品进口报关做网站,创建小程序的流程,网站全屏大图代码,二级域名网站查询From 产品经理#xff1a; 这是一份姗姗来迟的关于OceanBase并行执行的系统化产品文档。 自2019年起#xff0c;并行执行功能已被许多客户应用于多种场景之中#xff0c;其重要性日益凸显。然而#xff0c;遗憾的是#xff0c;我们始终未能提供一份详尽的用户使用文档 这是一份姗姗来迟的关于OceanBase并行执行的系统化产品文档。 自2019年起并行执行功能已被许多客户应用于多种场景之中其重要性日益凸显。然而遗憾的是我们始终未能提供一份详尽的用户使用文档这无疑给业务团队在运用并行执行功能时带来了诸多困扰。今日我们决心弥补这一缺憾。 关于并行执行的内容我们将通过七篇博客系列进行详尽的解析而本文正是这一系列的第一篇。 并行执行是指单个SQL语句能够同时利用多个 CPU 和 I/O 资源的能力。本文将深入探讨并行执行的工作机制同时阐明如何在OceanBase数据库中对其进行控制、管理与监控。
1 并行执行概念
并行执行是指将一个 SQL 查询任务分成多个子任务并允许这些子任务在多个处理器上同时运行以提高整个查询任务的执行效率。在现代计算机系统中多核处理器、多线程和高速网络连接广泛应用这使得并行执行成为了一种可行的高效率查询技术。
并行执行能够极大降低计算密集型大查询的响应时间被广泛应用在离线数据仓库、实时报表、在线大数据分析等业务场景同时还应用在批量导数快速构建索引表等领域。
以下场景会从并行执行中获益
大表扫描大表连接大数据量排序聚合大表 DDL 操作如修改主键、改变列类型建索引等从已有大数据建表Create Table As Select批量插入、删除、更新
本节包含如下内容
什么场景适用并行执行什么场景不适用并行执行硬件要求并行执行工作原理并行执行工作线程通过均衡负载来优化性能 1.1 什么场景适用并行执行
并行执行通过充分利用多个 CPU 和 IO 资源以达到降低 SQL 执行时间的目的。
当满足下列条件时使用并行执行会优于串行执行
访问的数据量大SQL 并发低要求低延迟有充足的硬件资源
并行执行用多个处理器协同并发处理同一任务在这样的系统中会有收益
多处理器系统SMPs、集群IO 带宽足够内存富余可用于处理内存密集型操作如排序、建 hash 表等系统负载不高或有峰谷特征如系统负载一般在 30% 以下
如果你的系统不满足上述特征那么并行执行可能不会带来显著收益。在一些高负载小内存或 IO 能力弱的系统里并行执行甚至会带来负面效果。
并行执行不仅适用于离线数据仓库、实时报表、在线大数据分析等分析型系统而且在 OLTP 领域也能发挥作用可用于加速 DDL 操作、以及数据跑批工作等。但是对于 OLTP 系统中的普通 SELECT 和 DML 语句并行执行并不适用。 1.2 什么场景不适用并行执行
串行执行使用单个线程来执行数据库操作在下面这些场景下使用串行执行优于并行执行
Query 访问的数据量很小高并发Query 执行时间小于 100 毫秒 并行执行一般不适用于如下场景
系统中的典型 SQL 执行时间都在毫秒级。并行查询本身有毫秒级的调度开销对于短查询来说并行执行带来的收益完全会被调度开销所抵消。系统负载本就很高。并行执行的设计目标就是去充分利用系统的空余资源如果系统本身已经没有空余资源那么并行执行并不能带来额外收益相反还会影响系统整体性能。 1.3 硬件要求
并行执行对硬件没有特殊要求。需要注意的是CPU 核数、内存大小、存储 I/O 性能、网络带宽都会影响并行执行性能其中任意一项成为瓶颈都会拖累并行执行性能。 1.4 并行执行工作原理
并行执行将一个 SQL 查询任务分解成多个子任务并调度这些子任务到多个处理器上运行。
本节包含如下内容
SQL 语句的并行执行生产者消费者流水线模型并行的粒度生产者和消费者之间的数据分发方式生产者和消费者之间的数据传输机制 1.4.1 SQL 语句的并行执行
当一个 SQL 被解析为并行执行计划后会按照下面的步骤执行
SQL 主线程接收、解析SQL的线程根据计划形态预约并行执行需要的线程资源。这些线程资源可能来自集群中的多台机器。SQL 主线程打开并行调度算子PX COORDINATOR。并行调度算子解析计划将它们切分成多个操作步骤按照自底向上的顺序调度执行这些操作。每个操作都会尽可能并行执行。当所有操作并行执行完成后并行调度算子会接收计算结果并将结果吐给它的上层算子如 Aggregate 算子串行完成剩余不可并行的计算如最终的 SUM 计算。 1.4.2 生产者-消费者流水线模型
并行执行使用生产者-消费者模型来进行流水执行。并行调度算子解析计划将它们切分成多个操作步骤每个操作步骤称之为一个 DFOData Flow Operation。
一般情况下并行调度算子在同一时刻会启动两个 DFODFO 之间会以生产者-消费者的模式连接起来这称为 DFO 间的并行执行。每个 DFO 会使用一组线程来执行这称为 DFO 内的并行执行这个 DFO 使用的线程数称为 DOPDegree Of Parallisim。
上一阶段的消费者 DFO 会成为下一阶段的生产者 DFO。在并行调度算子的协调下会同时启动消费者 DFO 和生产者 DFO。
下图中
1DFO A 生成的数据会立即传输给DFO B 进行计算
2DFO B 完成计算后会将数据暂存在当前线程中等待它的上层 DFO C 启动
3当 DFO B 收到 DFO C 启动完成通知后会将自己的角色转变成生产者开始向 DFO C 传输数据DFO C 收到数据后开始计算。 考虑下面的查询
create table game (round int primary key, team varchar(10), score int)partition by hash(round) partitions 3;insert into game values (1, CN, 4), (2, CN, 5), (3, JP, 3);
insert into game values (4, CN, 4), (5, US, 4), (6, JP, 4);select /* parallel(3) */ team, sum(score) total from game group by team;
查询语句对应的执行计划
OceanBase(admintest)explain select /* parallel(3) */ team, sum(score) total from game group by team;
---------------------------------------------------------------------------------------------------------
| Query Plan |
---------------------------------------------------------------------------------------------------------
| |
| |ID|OPERATOR |NAME |EST.ROWS|EST.TIME(us)| |
| ----------------------------------------------------------------- |
| |0 |PX COORDINATOR | |1 |4 | |
| |1 | EXCHANGE OUT DISTR |:EX10001|1 |4 | |
| |2 | HASH GROUP BY | |1 |4 | |
| |3 | EXCHANGE IN DISTR | |3 |3 | |
| |4 | EXCHANGE OUT DISTR (HASH)|:EX10000|3 |3 | |
| |5 | HASH GROUP BY | |3 |2 | |
| |6 | PX BLOCK ITERATOR | |1 |2 | |
| |7 | TABLE SCAN |game |1 |2 | |
| |
| Outputs filters: |
| ------------------------------------- |
| 0 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset256 |
| 1 - output([INTERNAL_FUNCTION(game.team, T_FUN_SUM(T_FUN_SUM(game.score)))]), filter(nil), rowset256 |
| dop3 |
| 2 - output([game.team], [T_FUN_SUM(T_FUN_SUM(game.score))]), filter(nil), rowset256 |
| group([game.team]), agg_func([T_FUN_SUM(T_FUN_SUM(game.score))]) |
| 3 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset256 |
| 4 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset256 |
| (#keys1, [game.team]), dop3 |
| 5 - output([game.team], [T_FUN_SUM(game.score)]), filter(nil), rowset256 |
| group([game.team]), agg_func([T_FUN_SUM(game.score)]) |
| 6 - output([game.team], [game.score]), filter(nil), rowset256 |
| 7 - output([game.team], [game.score]), filter(nil), rowset256 |
| access([game.team], [game.score]), partitions(p[0-2]) |
| is_index_backfalse, is_global_indexfalse, |
| range_key([game.round]), range(MIN ; MAX)always true |
---------------------------------------------------------------------------------------------------------
29 rows in set (0.003 sec)
select语句的执行计划会首先对 game 表做全表扫描按照 team 做分组求和然后算出每个 team 的总分数。这个查询的执行示意图如下 从图中可以看到这个查询实际使用了 6 个线程。
第一步前三个线程负责 game 表扫描并且在每个线程内对 game.team 数据做预聚合第二步后三个线程负责对预聚合的数据做最终聚合第三步最终聚合结果发给并行调度器由它返回给客户端。
第一步的数据发给第二步时需要用 game.team 字段做 hash决定将预聚合数据发给哪个线程。 1.4.3 并行的粒度
并行数据扫描的基本工作单元称为 granule。OceanBase 将表扫描工作划分成多个 granule每个 granule 描述了一段扫描任务的范围。因为 granule 不会跨表分区 所以每段扫描任务一定是位于一个分区内。
根据 granule 的粒度特征可以划分为两类
Partition Granule
Partition granule 描述的范围是一整个分区扫描任务涉及到多少个分区就会划分出多少个 partition granule。这里的分区既可以是主表分区也可以是索引分区。
Partition Granule 最常用的应用场景是 partition wise join两张表里对应的分区通过 partition granule 可以确保被同一个工作线程处理。
Block Granule
Block granule 描述的范围是一个分区中的一段连续数据。数据扫描场景里一般都是使用 block granule 划分数据。每个分区都会被划分为若干个 block这些 block 再以一定的规则串联起来形成一个任务队列供并行工作线程消费。 在给定并行度的情况下为了确保扫描任务的均衡优化器会自动选择将数据划分成分区粒度Partition Granule或块粒度Block Granule。如果选择了 Block Granule并行执行框架会在运行时决策 block 的划分总体原则是确保一个 block 既不会太大也不会太小。太大可能导致数据倾斜让部分线程少干活太小会导致频繁的扫描切换开销。
划分好分区粒度后每个粒度对应一个扫描任务。Table Scan 扫描算子会一个接一个地处理这些扫描任务处理完一个之后接着处理下一个直到所有任务处理完毕。 1.4.4 生产者和消费者之间的数据分发方式
数据分发方式指的是数据从一组并行执行工作线程生产者发送给另一组消费者时使用的方法。优化器会使用一系列的优化策略决策使用哪种数据重分布方式以达到最优的性能。
并行执行中的常见数据分发方式包括
Hash Distribution
使用 Hash distribution 发送数据时生产者根据 distribution key 对数据行计算 hash 值并取模算出发给哪个消费者工作线程。大部分情况下使用 hash distribution 能将数据较为均匀的分发给多个消费者线程。
Pkey Distribution
使用 Pkey Distribution 时生产者计算出数据行对应的目标表所在分区然后将行数据发给处理这个分区的消费者线程。
Pkey Distribution 常用于 partitial partitions wise join 场景。在 partitial partitions wise join 场景下消费者侧的数据不需要做重分布就可以和生产者侧的数据做 Partition Wise Join。这种方式可以减少网络通信量提升性能。
Pkey Hash Distribution
使用 Pkey Hash Distribution 时生产者首先需要计算出数据行对应的目标表所在的分区。然后根据 distribution key 对数据行进行 hash 计算以便决定将其发给哪一个消费者线程来处理。
Pkey Hash Distribution 常常应用于 Parallel DML 场景中。在这种场景下一个分区可以被多个线程并发更新因此需要使用 Pkey Hash Distribution 来确保相同值的数据行被同一个线程处理不同值的数据行尽可能均分到多个线程处理。
Broadcast Distribution
使用 broadcast distribution 时生产者将每一个数据行发送消费者端的每一个线程使得消费者端每一个线程都拥有全量的生产者端数据。
Broadcast distribution 常用于将小表数据复制到所有执行 join 的节点然后做本地 join 操作。这种方式可以减少网络通信量。
Broadcast to Host Distribution简称 BC2HOST
使用 broadcast to host distribution 时生产者将每一个数据行发送消费者端的每一个节点上使得消费者端每一个节点都拥有全量的生产者端数据。然后节点里的消费者线程协同处理这份数据。
Broadcast to host distribution 常用于 NESTED LOOP JOIN、SHARED HASH JOIN 场景。NESTED LOOP JOIN 场景里消费端的每个线程会从共享数据里取一部分数据行作为驱动数据去目标表里做 join 操作SHARED HASH JOIN 场景里消费端的每个线程会基于共享数据协同构建 hash 表避免每个线程独立构建相同 hash 表导致的不必要开销。
Range Distribution
使用 Range distribution 时生产者将数据按照 range 范围做划分让不同消费者线程处理不同范围的数据。
Range distribution 常用于排序场景各个消费者线程只需排序好发给自己的数据数据就能在全局范围内有序。
Random Distribution
使用 Random distribution 时生产者将数据随机打散发给消费者线程使得每个消费者线程处理的数据数量几乎一致从而达到均衡负载的目的。
Random distribution 常用于多线程并行 UNION ALL 场景该场景只要求数据打散负载均衡数据之间无其它关联约束。
Hybrid Hash Distribution
Hybrid hash distribtuion 用于自适应的 join 算法。结合收集的统计信息OceanBase 提供了一组配置项来定义常规值和高频值。Hybrid hash distribtuion 方法将 join 两侧的常规值做 hash 分布左侧的高频值使用 broadcast 分布右侧的高频值使用 random 分布。 1.4.5 生产者和消费者之间的数据传输机制
并行调度算子在同一时刻会启动两个 DFODFO 之间会以生产者-消费者的模式连接起来并行执行。为了方便在生产者和消费者之间传输数据需要创建一个传输网络。
例如生产者 DFO 以 DOP 2 来做数据扫描消费者 DFO 以 DOP 3 来做数据聚合计算每个生产者线程都会创建 3 个虚拟链接去连接消费者线程总计会创建 6 个虚拟链接。如下图 生产者和消费者之间创建的虚拟传输网络被称为数据传输层Data Transfer Layer简称 DTL。OceanBase 并行执行框架中所有的控制消息和行数据都通过 DTL 进行收发。每个工作线程可以对外建立数千个虚拟链接具有高度的可扩展性。除此之外DTL 还具有数据缓冲、批量数据发送和自动流量控制等能力。
当 DTL 链接的两端位于同一个节点时DTL 会通过内存拷贝的方式来传递消息当 DTL 链接的两端位于不同节点时DTL 会通过网络通信的方式来传递消息。 1.5 并行执行工作线程
一个并行查询会使用两类线程1个主线程若干个并行工作线程。其中主线程和普通的 TP 查询使用的线程没有任何区别来自普通工作线程池并行工作线程来自专用线程池。
OceanBase 使用专用线程池模型来分配并行工作线程。每个租户在其所属的各个节点里都有一个租户专属的并行执行线程池并行查询工作线程都通过这个线程池来分配。
并行调度算子在调度每个 DFO 之前会去线程池中申请线程资源。当 DFO 执行完成时会立即源释放线程资源。
线程池的初始大小为 0按需增长没有上限。为了避免空闲线程数过多线程池引入自动回收机制。对于任意线程
如果空闲时间超过 10 分钟并且线程池中剩余线程数大于 8 个则被回收销毁如果空闲时间超过 60 分钟则无条件销毁
虽然线程池的大小没有上限但是通过下面两个机制能在绝大多数场景里形成事实上限
并行执行开始执行前需要通过 Admission 模块预约线程资源预约成功后才能投入执行。通过这个机制能限制并发查询数量。Admission 模块的详细内容参考本文中 《3 并发控制与排队》一节。查询每次从线程池申请线程时单次申请的线程数量不会超过 N N 等于租户 UNIT 的 MIN_CPU 乘以 px_workers_per_cpu_quota如果超过则最多只分配 N 个线程。px_workers_per_cpu_quota 是租户级配置项默认值为 10。例如一个 DFO 的 DOP 100它需要从 A 节点申请 30 个线程从 B 节点申请 70 个线程UNIT 的 MIN_CPU 4px_workers_per_cpu_quota 10那么 N 4 * 10 40。最终这个 DFO 在 A 节点上实际申请到 30 个线程B 节点上实际申请到 40 个线程它的实际 DOP 为 70。 1.6 通过均衡负载来优化性能
为了达到最优性能所有工作线程分到的工作任务应该尽量相等。
SQL 使用 Block Granule 划分任务的时候工作任务会动态地分配到工作线程之间。这样可以最小化工作负载不均衡问题即一些工作线程的工作量不会明显超过其它工作线程。SQL 使用 Partition Granule 划分任务时可以通过让任务数是工作线程数的整数倍来优化性能。这适用于 Partition Wise Join 和并行 DML 场景。
举个例子假设一个表有 16 个分区每个分区的数据量差不多。你可以使用 16 工作线程DOP 等于 16以大约十六分之一的时间完成工作你也可以使用五个工作线程以五分之一的时间完成工作或使用两个线程以一半的时间完成工作。
但是如果你使用 15 个线程来处理 16 个分区则第一个线程完成一个分区的工作后就开始处理第 16 个分区。而其他线程完成工作后它们变为空闲状态。当每个分区的数据量差不多时这种配置会导致性能不优当每个分区的数据量有所差异时实际性能则会因情况而异。
类似地假设你使用 6 个线程来处理 16 个分区每个分区的数据量差不多。在这种情况下每个线程在完成其第一个分区后会处理第二个分区但只有四个线程会处理第三个分区而其他两个线程会保持空闲。
一般来说不能假设在给定数量的分区N和给定数量的工作线程P上执行并行操作所花费的时间等于 N 除以 P。这个公式没有考虑到一些线程可能需要等待其他线程完成最后的分区。但是通过选择适当的 DOP可以最小化工作负载不均衡问题并优化性能。
