性能竞赛优秀项目 | 分得干脆、合得高效，用 Shuffle 优化 TiDB 算子-低调大师

性能竞赛优秀项目 | 分得干脆、合得高效，用 Shuffle 优化 TiDB 算子

2020-12-28 525

作者介绍：黄建博，云计算领域技术开发工程师；金灵， Shopee 软件研发工程师。

他们的队伍 huang-b 在性能竞赛中斩获一等奖，本文将介绍 Shuffle 优化 TiDB 算子项目的设计与实践过程。

在我们往常的印象中，分与合是一对矛盾的概念，但是这次比赛留给我们队伍一个很深刻的印象是，分与合是一对互相促进的矛盾，只有干净利落地分解，才能高效地合并。这种印象一方面来自于我们的参赛思路，我们选择的方向是使用 Shuffle 操作将算子的数据源拆分为多个独立的分区，然后通过并行计算来提升整体的吞吐，优化的过程就是寻找更合适的分解方式，追求更好的扩展性和计算性能，相关技术细节在正文中有详细介绍。另一方面，则是来自于我们的参赛体验。

我们的队伍有两位成员一位顾问，分散在国内三座不同的城市，比赛从开始到结束，我们没有机会线下交流过，全都是以 slack 和文档的形式进行合作，我们开玩笑说，在做一个分布式的比赛的同时，我们也是一支分布式的队伍。在这样一种受限的条件下，有两个因素对队伍的高效合作起了关键作用：其一是我们的顾问对任务做了干净利落的拆分，同样的思路应用到两个不同类型的算子上，使得我们可以花开两朵，各表一枝；其二就是 TiDB 整体高内聚低耦合的设计，从纵向的角度讲，就是 TiDB 清晰的分层设计使得我们的优化可以只关注解析器和执行器层面，而不必深入更底层的 TiKV 存储，从横向的角度讲，就是 TiDB 对分治和多态的充分实践使得我们可以只关注被优化的算子，而不必担心对其它算子产生副作用。

技术背景

我们的优化思路是使用 Shuffle 算子来实现 MergeJoin 算子和 StreamAggregation 算子的并行化。Shuffle 算子最先在 PR https://github.com/pingcap/tidb/pull/14238 中引入，用于并行化 Window 算子。图 1 展示的就是 Window 算子的并行化过程。

图中左侧是串行的 Window 算子，因为 Window 算子要求输入数据有序，所以在数据源和 Window 之间通常有一个 Sort 算子。图中右侧展示的是对应的 Shuffle 算子，为了完成并行计算，Window 算子和 Sort 算子都被复制了多份，每一份与一 ShuffleWorker 相对应，从数据源流入的数据由 Splitter 按哈希值拆分为独立的数据分区，发往不同的 ShuffleWorker ，最终各个 Window 算子的结果汇总后输出。图中所画箭头就是数据流动的方向，其中数据的分发和结果的汇总是通过 go channel 来实现的，其它数据流动都是父节点通过调用子节点的 Next 方法来获取的。图中虚线表示启动的协程，每个 ShuffleWorker 都会启动一个协程来完成自身的运算，同时 Splitter 也会启动一个协程来完成数据的分发。

题目链接：

1. ShuffleMergeJoin:https://github.com/pingcap/tidb/issues/14441

2. ShuffleStreamAgg:https://github.com/pingcap/tidb/issues/20651

图 1 Window 算子并行化

ShuffleMergeJoin

扩展 Shuffle 算子

对 MergeJoin 做并行优化，是不是简单套用 ShuffleWindow 的框架就可以了呢？不是的，MergeJoin 算子与上文的 Window 算子不同，MergeJoin 需要两个数据源。那现在的 Shuffle 实现能不能让每个并行算子对应两个 ShuffleWorker ，进而对应两个数据源呢？答案也是不可以，因为前文提到的 Shuffle 实现把数据分区和计算并行这两个功能过度耦合在一起了，这种过度耦合使得它无法支持两个数据源。下面我们对这个问题作具体说明。

过度耦合指的是 ShuffleWorker 充当的角色太多，它既是数据流动的一环，同时也是计算并行的基本单元，于是带来了这两个问题：

1. 因为 ShuffleWorker 时数据分区，所以并行后的每个 MergeJoin 需要两个 ShuffleWorker 来接收来自两个数据源的数据，但是 ShuffleWorker 同时又是计算并行的基本单元，于是有 n 个 MergeJoin 算子就会出现 2n 个协程，同一个 MergeJoin 算子的两个协程还会出现数据竞争。

2. 控制逻辑复杂， ShuffleWorker 作为数据的一个分区，它必须作为 Sort 算子的子节点，而它作为计算并行的基本单元，又必须在协程中调用 Window 算子的 Next 方法来完成计算，所以在原来的实现往 ShuffleWorker 里放了个指向 Window 算子的指针，这样的设计一方面存在破坏执行树有向无环特性的隐患，另一方面也降低了代码的可读性。

当然，第一个问题我们可以通过在 ShuffleWorker 中增加一个布尔变量来解决：同一个 MergeJoin 对应的两个 ShuffleWorker 一个为 true，一个为 false，只有为 true 的那个才会启动协程。可是这个方法无疑会使上面提到的复杂的逻辑更加复杂。

**我们提出的解决方案是把数据分区和计算并行解耦。**如图 2 所示：计算并行还是由 ShuffleWorker 负责，但是它不再是数据流动过程中的一环，它原来在数据流动过程中的位置由 ShuffleReceiver 来代替。MergeJoin 是 ShuffleWorker 的一个成员，每个 ShuffleWorker 对应一个协程，在协程中调用 MergeJoin 的 Next 方法，并将结果发送给汇总算子，这样上文中提到的两个问题都得到了解决。

图 2 拓展后的 Shuffle 算子

实现与效果

在实现中我们考虑两个场景：其一是数据源本身无序的情况，这种情况下数据进入 MergeJoin 之前要先经过 Sort 节点；另一是数据源本身有序的情况，这种情况下数据进入 MergeJoin 之前无需排序。

图 3 数据源无序情况下的 ShuffleMergeJoin

图 3 展示的就是数据源无序情况下 MergeJoin 的并行化过程，这种情况 MergeJoin 和 Sort 算子的计算开销都可以分摊到多个协程上。启动 2 个 worker 的优化效果如表 1 所示，我们在不同规模的数据源上都做了测试，表中前两列是两个数据源的行数，表中的后两列是串行和并行版本的运行性能，单位是 ns/op，越小性能越高。从表中可以看出， Shuffle 是可以明显加速 MergeJoin 运算的，并且数据量越大的情况下加速效果越好（因为并行化是会引入管道、协程等额外开销的，比较大的数据量才能保证并行化的收益大于开销）。在我们的几个测试案例中，效果最好的情况下 2 个 worker 的运算时间仅为串行版本的 56.5% 。

表 1 ShuffleMergeJoin 优化效果

图 4 展示的是数据源有序情况下 MergeJoin 的并行化过程，区别就是数据不再经过 Sort 算子。这种情况下计算的负载本身比较轻量，相比之下根据哈希值来分发数据的 Splitter 就成为了系统的性能瓶颈，并行化以后性能提升并不明显。

图 4 数据源有序情况下的 ShuffleMergeJoin

ShuffleStreamAggregation

聚合运算是 SQL 语句必不可少的一部分，无论是 OLTP 还是 OLAP 场景，聚合都是经常被使用到的算子。

从系统实现层面来看，聚合的实现一般有两种，第一种是基于 Hash 的方法，该方法通过构建 Hash table，维护每一个被聚合元素的值，计算得到最后的结果值。另外一种，则是基于有序数据流的方法，该方法要求输入数据源必须是有序的，然后通过遍历有序的数据流，并在同时维护相应的聚合值，即可得到最后的计算结果。

一般来说，基于 Hash 的方法具有更快的计算速度，但是它需要维护一个 Hash table，内存空间使用成本较高，当被聚合 key 的可能取值个数非常大的时候，那么相应 Hash table 中的元素个数也会非常多，对内存是个不小的考验，存在爆内存的风险，这反而导致计算不能正常地被完成。而基于有序数据流方法的聚合运算实现方式，无需随时都在内存中维护所有的被聚合 key 的值，因此内存使用量相对较小，但是它的运行速度相对而言更慢一点，且更为严格地要求输入数据必须是有序的。如果可以提升基于有序数据流方法的聚合算子的运行速度，那么该方法将会更加适用于大数据量的情况。因此我们选择对基于有序数据流方法的聚合运算实现方法，即 Stream Aggregation 进行并行加速，以提升该算子的整体运行速度。

实现与效果

在具体的实现过程中，我们利用了之前由其他社区贡献者提供的 Shuffle 算子，在 StreamAggregation 算子外围，将输入数据分割成多个有序的输入数据流，分别输入到多个 StreamAggregation 算子当中，然后通过简单的整合，得到最后的计算结果。简单地说，初始输入数据源 DataSource,首先会经过 Shuffle 算子，被分割成多个 Partition，且每个 Partition 都是其内部有序的，然后每个 Partition 分别被作为一 StreamAggregation 算子的输入，生成部分结果，最后通过对相同 key 的元素进行整合，即可得到最后的整体计算结果。

此处需要考虑 DataSource 是否有序的情况，如果 DataSource 在被聚合 key 上是无序的，比如普通的 PhysicalTableReader 算子，或其他算子的输出，那么需要在被分割之前，使得其有序，因此需要在其上添加一个 Sort 算子（如图 5 所示）。

图 5 数据源无序情况下的 ShuffleStreamAggregation

针对这种场景，我们的方法最后取得了非常明显的性能提升（如表 2 所示）。分析认为，非并行的情况下，Sort 作用在整个 DataSource 之上，而并行化的版本是作用在每个不同的 Partition 之上，输入相对较小，且并行执行，因此性能提升较大。

表 2 ShuffleStreamAggregation 优化效果

另外，还需要考虑 DataSource 在被聚合 key 上是有序的情况，比如下面的 SQL 语句，被聚合 key 为 b，且输入数据源 t 上刚好有由 b 创建的索引，因此在具体的计算过程中， DataSource 是基于 b 的 PhysicalIndexTableReader ，那么我们就无序引入 Sort 算子，直接将输入分割成多个 Partition，然后经过图 6 所示的计算过程即可得到结果。

create table t(a int, b int, key b(b)); 
select /*+ stream_agg() */ count(a) from t group by b;

图 6 数据源有序情况下的 ShuffleStreamAggregation

通过 Benchmark 的结果表明，在该情况下，目前的基于 Shuffle 的实现，运行速度并没有得到提升，反而有所下降，我们粗浅地认为，当前的 Shuffle 实现方式是瓶颈点，是后续需要被解决的重点。

RangeSplitter

上面说了， Shuffle 算子会把数据输入分割成多个 Partition，最开始的时候只有基于 Hash 方法的 Splitter 被实现，该实现对输入数据是否有序并不做要求。对于数据源有序的情况，尽管该方法依旧适用，但是使用基于 Range 的方法对数据源进行分割，是一个更为自然的方式，因为被聚合 key 相同的多行数据，必然是紧挨在以前的，如果可以直接找到这一块数据的起始点和结束点，整体一次性分割，则无需构建 HashTable，也不用调用开销更为明显的 HashFunction，使得Partiton 过程的开销更小。基于这一思路，我们实现了 PartitionRangeSplitter，该方法的计算原理是，将紧挨一起的相同聚合 key 的多行数据，批量地分发到一个 worker 之上。相较于基于 Hash 方法的 Partitioner 而言，基于 Range 方法的实现方式的开销更小，在同时处理有序输入数据源的情况下，使用 RangePartitioner 能比使用 HashPartitioner 快上一倍的速度（详见表 3），由此证明了该算子更加适用于数据源有序的情况。

表 3 RangeSplitter 与 HashSplitter 性能比较

鸿蒙应用开发入门（五）：日志HiLog的使用

目录： 5.1 概述 5.2 HiLog使用说明 5.3. 写demo练习 5.1 概述做一个Java攻城师，我们除了关心系统的架构这种high level的问题，还需要了解一些语言的陷阱，异常的处理, 以及日志的输出，这些"鸡毛蒜皮"的细节。我们需要通过打印一条条日志来掌握程序运行的状态，下面我们就来讲解鸿蒙系统中的HiLog日志工具的具体使用方法。 5.2 HiLog使用说明 1. 使用HiLog前必须在HiLog的一个辅助类HiLogLabel中定义日志类型、服务域和标记。一般我们把它定义为常量放在类的最上面： static final HiLogLabel label = new HiLogLabel(HiLog.LOG_APP, 0x00201, "MY_TAG"); 上面有三个参数： 1）日志类型，我们的应用一般取一个常量值：HiLog.LOG_APP，表示是第三方应用。 2）服务域，16进制整数形式，取值范围是0x0 ~ 0xFFFFF。一般情况下，我们建议把这5个16进制数分成两组，前面三个数表示应用中的模块编号，后面两个表示模块中的类的编号。 3）一个字符...

2020-12-28

556

2030年夏季，周日，天气很好，张三利用难得的一周唯一一天休息日，陪着孩子来上英语课，女儿5岁，在中关村上英语课，他则在教室旁边的星巴克喝咖啡，星巴克这个时候大多是孩子家长在这里等待孩子下课。张三期望孩子英语学好，孩子在将来能少把时间花在英语上以学好数理化，他老婆则不是这个想法，希望能在小升初的时候，能通过英语被点招到更好的学校，尽管这个希望很渺茫，政策总是变化。英语学费挺高，每三个月需要1万，但张三还是挺舍得对自己女儿花这钱，他对自己信心满满，他现在处于事业上升期，在国内一个著名的平台电商公司上班，不到29岁，年薪总包已经50万了。加上他老婆挣的钱，应付孩子学前教育和课外班，房供绰绰有余。张三在网关组里担任leader，他憧憬自己未来职位提升，年薪百万，能积分落户在北京。张三旁边有个中年人，大约45岁左右，带着眼镜，胡子邋遢，白头发很多，他也经常在这个店里，张三知道中年人手里的捧着的咖啡不是他自己的，这个中年人他每次来都会很早来店里找个位置坐好，如果他看到其他客人有喝剩下的咖啡或者吃剩下的三明治，他都会坐过去，然后捡起来吃。店员似乎也不管他，他并没有影响任何人，他偶尔他也会掏钱...

2020-12-28

521

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。