search

mrdrivingduck / paper-outline

🔍 To record the papers I have read.
24 stars 0 forks source link

Citus: Distributed PostgreSQL for Data-Intensive Applications #8

Open mrdrivingduck opened 3 years ago

mrdrivingduck commented 3 years ago

3448016.3457551.pdf - SIGMOD 2021

Citus 是一个 Microsoft 开发的开源分布式数据库引擎，被实现为一个数据库扩展。Citus 使得用户可以在若干个单机 PostgreSQL 数据库服务器组成的集群上进行分布式的数据存储、查询、事务处理。其开发过程由四类常见的 workload 启发。本文详细介绍了这四类常见 workload 的模式，并解释了 Citus 如何解决这四类 workload 的问题。

mrdrivingduck commented 3 years ago

1. Introduction

过去十年该团队一致致力于开发开源的 PostgreSQL 扩展 (插件) - Citus，它能够将 PostgreSQL 成为一个 D (Distributed) DBMS。其目标是解决 PG 生态中的扩展性需求。

新的 DDBMS 如果想要与已有的单机关系型 DBMS 兼容，遵循了以下三种方法之一：

  1. 从头开始设计数据库引擎，并提供 SQL 兼容性转换层
  2. 从开源 DBMS 中 fork 一份，并在上面开发新特性
  3. 以中间件的形式在应用和 DBMS 中间提供特性

对于上面的任何一种方法，跟踪正在开发的核心工程的工作量都很大。

Citus 是一个通过使用 PostgreSQL 的扩展 API 来提供功能的分布式数据库。扩展 API 提供了对 PostgreSQL 行为的足够控制权：

构建一个与 PostgreSQL 100% 兼容的分布式引擎是不可能的，也不是所有的 workload 都能够从 scale out 中受益。本文介绍了可以从 PG scale out 中受益的四种 workload 模式，每种模式都需要不同的数据库功能组合。

mrdrivingduck commented 3 years ago

2. Workload Requirements

在实际中，延时、吞吐量、数据规模等要求随着 workload 的不同而极度不同，因此不同的 workload pattern 需要不同的分布式数据库的能力组合。以下是四种 workload 所需的特性表格：

image

2.1 Multi-tenant

多租户应用程序，通过部署单个后端，服务大量相对独立的租户。典型应用为 SaaS。增加一个新租户的代价极其小，但由于租户数量大，工作集的大小会很大。

扩展多租户应用的传统方式为 手动分片。每一个租户的数据被放到租户自己的数据库中，另外应用需要追踪每一个数据库被放置的位置。

另一种方法是使用租户 ID 列来进行分片。包含租户相关数据表需要有一个租户 ID 列，由该列对数据进行分区和 co-located。Co-located 保证同一个租户 ID 总是放置在同一台服务器上，从而保证连接和外键总是在同一台服务器上。DBMS 需要有能力将任意复杂的 SQL 查询根据租户 ID 过滤后路由到相应的服务器上。另外应用还可能执行跨租户的事务处理和分析。

2.2 Real-time Analytics

交互式分析，或以很小的延时搜素大量数据。典型数据为事件数据和时序数据。数据库需要具备较高的写吞吐来保证追上数据产生的速度，同时还要能够执行大量的分析。应用可以使用索引、物化视图等变换方式来最小化响应时间，但数据库也需要增量更新这些信息。

PG 的 heap 表存储结构和 COPY 命令能够很好地进行快速数据获取，同时 MVCC 保证了分析查询能够与写入并发进行。但是 PG 的缺点是：大部分操作都是单线程的。为了扩展到实时分析 workload，DBMS 需要能够将表跨 server 分布，并且支持并行批量加载数据。分布式并行 DML (尤其是 INSERT...SELECT) 需要增量聚合大量数据到 rollup table 中。源表和 rollup table 的 co-locate 意味着并行 DML 能够快速完成。

2.3 High-performance CRUD

包含以相对独立的方式修改大量对象/文档的 workload，有对 key 进行修改的简单操作，也有更复杂的跨对象查询。PG 的局限在于 vacuum 的速度跟不上对象更新的速度；另一个局限在于 PG 的进程模型决定了它只能处理有限个数的空闲连接。

为了扩展高性能的 CRUD，表需要能够按照 key 来进行分布。另外，vacuum 也需要在多个 core 上并行。并行、分布式的 SELECT 和 DML 被用于在大量对象上进行扫描和分析。为了扩展连接数量，任何 server 都需要能够执行分布式查询。

2.4 Data warehousing

查询需要扫描大量数据。为了扩展数据仓库应用，扫描需要通过并行、分布式的 SELECT 和列式存储进行加速。分布式列需要以最大化 co-located 分布式连接数为原则进行选择，但数据库也要支持高效的 non-co-located 连接 (通过 reshuffling 或广播实现)。查询优化器需要以最小化网络带宽为原则决定 join order。

mrdrivingduck commented 3 years ago

3. Citus Architecture

Citus 使用 PostgreSQL 的扩展 API 来改变数据库的行为：

  1. Citus 将数据库对象 (自定义类型和函数) 复制到所有 server 上
  2. Citus 加入了两种新的表类型，用于扩展更多的 server

3.1 PostgreSQL extension APIs

一个 PG 扩展包含两部分：

当 PG 扩展被装载后，它能够通过设置特定的 hook 来改变 PG 的行为。Citus 使用了以下 hook：

通过这些 hook，Citus 可以截获任何包含 Citus 表的交互，并改变 PostgreSQL 的行为。

3.2 Citus architecture diagram

Citus 包含一个 coordinator 节点和 0 到 n 个 worker 节点。Coordinator 节点保存 metadata。当用户使用 Citus UDF 添加一个 worker 节点时，原节点隐式成为 coordinator 节点。Worker 节点保存实际存储数据的分片，当数据较少时，coordinator 也可以保存数据。设置单个 coordinator 的好处在于 PG 周边工具可以像使用单机数据库一样与 Citus 集群进行交互。

3.2.1 Scaling the coordinator node

Coordinator 有时可能会成为扩展瓶颈。为了解决这个瓶颈，Citus 可以将所有分区表的 metadata 都分布到所有的 worker node 上，客户端使用负载均衡机制直接连接到 worker node。执行 DDL 的客户端还是需要连接到 coordinator 上。

当集群内所有节点都可以服务请求时，Citus 在集群内创建连接并池化缓存。

3.3 Citus table types

Citus 特有的表以普通 PG 表的方式被创建，然后通过 Citus 的内置函数进行转换。转换后，Citus 将在 hook 中截获所有关于 Citus 表的操作。

3.3.1 Distributed tables

分布式表是一个在分布式列上进行 hash 分区的表。它被分成多个逻辑分片，每个分片内包含了连续范围的 hash value。Hash 分区的优势在于能够保证 co-location 和相对均衡的数据。

在创建分布式表后，分片以轮转的方式放置到 worker node 上。

3.3.2 Co-location

Citus 保证相同范围的 hash 值总是被存储在相同的 worker node 上。这样涉及到分布式列的关系型操作 (join / foreign keys) 可以在无需网络传输的情况下进行。在创建分布式表时，可以指定分布式列与另一个表的分布式列是 co-located 的。

3.3.3 Reference tables

这类表在 Citus 集群中会被复制到所有节点上。Distributed table 和 reference table 之间的 join 是通过把 distributed 表的每一个分片与 reference 的每一个本地备份进行 join 实现的。对于 reference table 的写操作将会被复制到所有节点上，读操作将会直接被 coordinator (或负载均衡地被 worker) 节点响应。

3.4 Data rebalancing

Citus 提供 shard rebalancer。默认情况下，rebalancer 将会工作直到分片数到达平均数；但用户也可以自行指定策略。

在 rebalancing 时，首先通过逻辑复制在目标节点上创建备份，原分片继续相应读写请求；当备份分片追上进度时，Citus 获取原分片上的写锁，等待复制完成，并更新 distributed table 的 metadata。之后，所有的新请求将会被路由到新的 worker node 上。中间的阻塞时间很短。

3.5 Distributed query planner

分布式 planner 将会产生带有 CustomScan 节点的查询计划，该节点内包含了分布式执行计划。一个分布式计划中包含：

Citus 对不同类型的查询有着不同的 planner：

对于每一个查询，Citus 都会从最低开销到最高开销遍历四个优化器。如果有一个计划可以产生，那么就使用它。

3.6 Distributed query executor

当 PG 执行器执行到 CustomScan 时，Citus 首先执行子计划，然后将执行流交给 adaptive executor。Adaptive executor 以 慢启动 的方式增大节点之间连接池的大小，并保证连接数不超过极限。

image

3.7 Distributed transactions

Citus 的分布式事务包含客户端在 coordinator 上发起的事务，以及执行器在一个或多个 worker 上发起的事务。如果事务只牵扯到一个节点，那么 Citus 会将责任下放到 worker node 上；否则 Citus 将使用 两阶段提交 (2PC) 来保证事务原子性，并实现分布式的死锁检测。

3.7.1 Single-node transactions

当处理单个 statement 的事务时，没有额外开销；当处理多个 statement 并且所有的 statement 都被路由到同一个 worker node 时，worker node 提供与单机 PG 相同的事务保障。

3.7.2 Two-phase commit protocol

对于需要写入多个节点的事务，执行器会在 worker node 上开启事务，并在 commit 时完成 2PC。PG 实现了事务准备状态的命令，Citus 使用这些命令来实现 2PC 协议。

当 coordinator 上的事务要提交时，通过 pre-commit 回调发送 prepare transaction 到所有开启事务的 worker node 上。如果成功，那么 coordinator 将写入一条 commit record 到 Citus metadata 以及本地事务中，保证 commit record 持久化。在 post-commit 和 abort 的回调中，以 best-effort 的原则来决定事务是否提交。

当一个或多个事务无法提交或中止时，Citus metadata 中的 commit record 被用于决定事务的结果。

Then...不懂了...

3.7.3 Distributed deadlocks

Citus 实现了分布式死锁检测，当检测到死锁时，中止事务。Citus 扩展了 PG 自带的死锁检测机制，算法运行在 coordinator 的守护进程中。该守护进程每两秒钟轮询所有 worker node 并收集 lock graph，然后在图中对参与同一个分布式事务的进程进行合并。如果图中出现环路，那么将会对最晚开始的分布式事务发送取消信号。

3.8 Specialized scaling logic

3.9 High Availability and backups

高可用在 server level 实现，使用了 PG 已有的复制机制。集群中每一个节点都有一个或多个 hot standby nodes，以同步或异步的方式回放 WAL。当节点失效时，集群编排器将一个 standby 接替，并更新集群 metadata、DNS 记录、虚拟 IP。

备份也是在 server level 实现的，通过周期性创建磁盘快照，或拷贝数据库目录同时不断归档 WAL。Citus 支持周期性创建一致性的恢复点，通过阻塞对 coordinator 上 commit record 表的写操作实现。