关于云原生:云原生数仓如何破解大规模集群的关联查询性能问题

3次阅读

共计 4445 个字符，预计需要花费 12 分钟才能阅读完成。

简介：AnalyticDB for PostgreSQL(以下简称 ADB PG)是一款 PB 级的 MPP 架构云原生数据仓库。本文从 ADB PG 架构设计的角度登程，探讨 Runtime Filter 在 ADB PG 中的实现计划，并介绍了基于 Bloom Filter 的 ADB PG Dynamic Join Filter 性能技术细节。

作者 | 宇毅
起源 | 阿里开发者公众号

近年来，数据库系统服务的数据量呈指数级增长，同时也面临解决的业务需要愈发简单、实时性要求越来越低等挑战。单机数据库系统曾经逐步不能满足古代的数据库服务要求，因而分布式数据库 / 数据仓库失去了越来越宽泛地使用。

在实时剖析（OLAP）畛域，分布式数据仓库能够充分发挥零碎的分布式特点，将简单的 OLAP 工作合成下发到零碎中的所有节点进行计算晋升剖析性能；分布式数据仓库也能够比拟不便地对系统节点进行扩容，应答用户业务数据量减少的需要。然而分布式数据仓库用户无奈防止的一个问题是：随着数据仓库集群规模增大，扩容带来的性价比愈发升高。

造成这种景象的一个起因是，表连贯（Join）作为数据库业务中最宽泛应用的算子之一，在分布式计算中依赖零碎节点间的数据交互；当分布式集群规模增大时，节点之间的数据交互代价会明显增加，这种状况下十分考验分布式系统的网络解决能力，并依赖用户的数据表设计和 SQL 编写能力以缓解数据交互压力。

针对这个问题，业界不同的分布式数据库系统提出了不同的 Join 运行时过滤 (Runtime Filter) 算法。AnalyticDB for PostgreSQL(以下简称 ADB PG)是一款 PB 级的 MPP 架构云原生数据仓库，同样也面临着上述问题的挑战。本文从 ADB PG 架构设计的角度登程，探讨 Runtime Filter 在 ADB PG 中的实现计划，并介绍了基于 Bloom Filter 的 ADB PG Dynamic Join Filter 性能技术细节。

ADB PG 基于开源我的项目 Greenplum 构建，在单机 PostgreSQL 的根底上进行扩大，将多个 PG 服务同时启动在单个或多个服务器上并组成集群，以分布式的模式提供数据库服务。

ADB PG 将每一个 PG 服务称为一个 Segment，并引入了 Slice 的概念。Slice 用于解决分布式系统中的网络结构，当数据库波及到 MPP 多阶段计算时，例如 Hash Join 左右表的 Join Key 不满足雷同的 Hash 散布，那么就须要对 Join Key 通过网络传输进行重散布，ADB PG 将网络传输的前后阶段切分为不同的 Slices。以下是一个 ADB PG 集群示意图。

在这种架构下如何解决大规模集群下表连贯 Join 的性能问题呢？业界解决这个问题的一个计划是引入网络代理节点，同一机器内的 Segment 将网络数据发送至本地代理节点，由代理节点与其它机器上的代理节点进行网络收发工作以缩小网络拥塞。该计划对 ADB PG 架构的挑战较大，且没有从根本上缩小 Join 的网络 Shuffle 开销。因而为了从 Join 本源上缩小 Join 计算的数据量，ADB PG 设计并实现了 Join Runtime Filter 计划。

Runtime FIlter 的目标是在 Join 计算前筛选掉一部分数据，须要一个 Filter 的实现“载体”。在联合 ADB PG 的架构设计、存储层和网络层的特点后，咱们抉择应用 Bloom Filter 作为 Runtime Filter 的实现模式。

Bloom Filter 是一种概率数据结构，通常被用于判断一个元素是否属于一个汇合。Bloom Filter 的长处是其空间效率十分高，计算性能通常也高；毛病是存在阳性误判率 false positive，然而不存在 false negative，即 Bloom Filter 判断一个元素是否属于汇合的后果不是单纯的 true or false，而是 ”possible true” or “false”。

上图是一个规范 Bloom Filter 的计算思路示意图，其中的 0、1 为 Bloom Filter 用于示意汇合信息的 bit array，即每一位用一个 bit 存储。上方 x,y,z 示意向 Bloom Filter 中插入的三个元素，别离应用 3 种 hash 算法计算 hash 值后在 bit array 中置位。而下方为判断元素 w 是否属于汇合，因为 3 个 hash 值中的某一位没有在 bit array 中被置位，能够必定的是 w 不属于汇合。

Bloom Filter 通常由以下几个参数形容：

m — Bloom Filter bit array 的大小 m bits
k — 应用的 hash 函数个数 k
p — 误判率
n — Bloom Filter 插入的元素个数

咱们省略推导过程，间接将各个参数的关系给出：

当 Bloom Filter 足够大时，能够简化为

在设计 Bloom Filter 时，n 和 m 咱们能够依据理论计算场景提前确定，上述公式能够视为自变量为 k，应变量为 p 的函数 p(k)，此函数通常在 k > 0 时通常不是枯燥的（由 n:m 确定）。因而 Bloom Filter 在设计时要思考如何确定 hash 函数 k 的个数以取得最小的误判率 p。依据上式能够计算失去当 p 为极小值时，对应 k 的值为：

Bloom Filter 的参数设计

如何将 Bloom Filter 利用至 ADB PG Join 过滤优化，咱们首先要设计抉择 Bloom Filter 的参数。对于 Bloom Filter 插入元素的个数 n，能够间接应用执行打算中取得的 Join 右表打算行数；而为了获得理想的过滤率，缩小误判率 p，ADB PG 应用了 PG 高版本 Bloom Filter 的思路，设计 Bloom FIlter 大小 Bytes 为 n 的 2 倍，即总体 n:m 达到 1:16。在这个设计下，能够计算失去最佳的 k 取值为 11，p(k)函数如下图所示，当 k = 11 时能够获得最小的 p = 0.046%

k = 11 意味着对于每一个元素，都须要计算 11 个 hash 值以插入到 Bloom Filter bit array 中，这对于 ADB PG 是无奈承受的，构建 Bloom Filter 的代价显著过大。在构建 Bloom Filter 时，ADB PG 会综合误判率、hash 计算等因素思考，抉择适合的 k 值。

在确定构建 Bloom Filter 的根本准则后，接下来就是工程实现问题。Bloom Filter 的工程实现非常简单高效，通常咱们能够间接应用 bitset 数组来建设 Bloom Filter，通过位操作实现 Bloom Filter 的插入和查找。下图为向一个 Bloom Filter bitset 数组中插入元素的计算示意图。

在实现 ADB PG Hash Join 的 Bloom Filter 设计后，接来下探讨如何将 Bloom Filter 利用至 Join 的 Runtime Filter 中。ADB PG 将基于 Bloom Filter 的 Runtime Filter 命名为 Dynamic Join Filter。

1 Dynamic Join Filter 的实现形式

因为 ADB PG 优化器通常会抉择将右表作为小表，左表作为大表，因而 ADB PG 将 Dynamic Join Filter 的设计特点为单向过滤的，即仅用于右表过滤左表，暂不思考左表过滤右表的模式；同时咱们也能够将 Dynamic Join Filter 灵便利用于 Hash Join 左表链路不同算子的过滤中。

因为 Hash Join 的模式不同，Dynamic Join Filter 的实现模式能够总结为 Local Join 和 MPP Join 两种模式，并依据 Runtime Filter 是否具备下推算子的能力做进一步辨别。

Local Join

Local Join 是指左右表的 Join Key 均满足雷同 Hash 散布，无需再 Shuffle 数据。此时 Hash、Hash Join 和左表 Scan 处于同一个 Slice 外部，即同一个过程中，咱们能够间接在过程空间内将 Bloom Filter 传递给左表 Scan 算子过滤输入。

MPP Join

MPP Join 是指左右表的 Join Key 均不满足雷同 Hash 散布，须要针对 Join Key Shuffle 数据。在前文介绍过，ADB PG 的 Hash Join 和 Hash 算子肯定处于同一个 Slice 外部，因而基于根本准则只须要思考左表 Shuffle 的状况，即左表在 Hash Join 前存在 Motion 的场景。

MPP Join 存在的另一种状况是，左表 Motion 下不是简略的 Scan，也没有关联信息将 Join Key 的 Bloom Filter 下推至 Scan。那么以缩小网络传输数据量为最初准则，将 Bloom Filter 过滤放在 Motion 前，缩小 Motion Sender 的数据。

2 Bloom Filter 网络传输

Dynamic Join Filter 在各个计算节点上建设了一个 Local Bloom Filter，每个计算节点须要收集所有其它节点的 Bloom Filter，并在本地组成残缺的 Bloom Filter 后能力开始过滤计算。咱们将 Bloom Filter 的收发分为两种模式：全量传输和位传输。在发送前咱们能够判断两种模式的数据量大小，并自适应抉择数据量小的模式。

Bloom Filter 全量传输

Bloom Filter 位传输

接下来咱们对 ADB PG Dynamic Join Filter 的性能体现测试。测试集群为 ADB PG 私有云搭建的实例，测试应用 TPC-H 1TB 测试集（scale = 10000），测试通过开启 \ 敞开 Dynamic Join Filter 性能比照执行性能。下图展现了 TPC- H 执行性能有差别的 Query 测试后果：

能够看到 Dynamic Join Filter 在 Q5、Q8、Q9 和 Q17 上均取得了较大的性能晋升，其中 Q17 的优化性能最佳，执行工夫 137s 优化至 8s。而 Q10 存在稍微的性能回退：10s 回退至 12s，起因在于 Q10 的 Join Key 是齐全匹配的，Dynamic Join Filter 无奈做到动静提前过滤，而优化器未能精确估算代价导致打算依然应用了 Dynamic Join Filter。此外 Q20 也因为优化器下推规定的的起因没有抉择 Dynamic Join Filter，实际上通过剖析 Q20 与 Q17 相似，比拟适宜应用 Dynamic Join Filter。为了解决这些问题，ADB PG 优化器相干性能仍在开发迭代中。

Dynamic Join Filter 依据 ADB PG 架构设计、存储层和网络层特点，应用 Bloom Filter 作为 Join Runtime Filter 的实现模式，在 TPC- H 测试中获得了显著的性能晋升成绩。将来咱们将从以下几个方面做进一步的开发和优化，晋升客户应用体验：