Netflix 发布 DBLog：一款通用的变化数据捕获框架

变化数据捕获 （CDC）允许从数据库实时捕获已提交的更改，并将这些更改传播到下游消费者。在需要保持多个异构数据存储同步（如 MySQL 和 ElasticSearch）的用例中，CDC 变得越来越流行，它解决了双写和分布式事务等传统技术存在的挑战。

本文最初发布于 Medium，由 InfoQ 中文站翻译并分享。

简介

变化数据捕获 （CDC）允许从数据库实时捕获已提交的更改，并将这些更改传播到下游消费者。在需要保持多个异构数据存储同步（如 MySQL 和 ElasticSearch）的用例中，CDC 变得越来越流行，它解决了双写和分布式事务等传统技术存在的挑战。

在 MySQL 和 PostgreSQL 这样的数据库中，事务日志是 CDC 事件的来源。由于事务日志的保留期通常有限，所以不能保证包含全部的更改历史。因此，需要转储来捕获源的全部状态。已有几个开源 CDC 项目，它们通常使用相同的底层库、数据库 API 和协议。尽管如此，我们发现了许多不能满足我们需求的限制，例如，在转储完成之前暂停日志事件的处理，缺少按需触发转储的能力，或者使用表级锁来阻止写流量的实现。

这是我们开发 DBLog 的动机，它在通用框架下提供日志和转储处理。它支持的数据库需要具有 MySQL、PostgreSQL、MariaDB 等系统中常见的一组特性。

DBLog 的部分功能如下：

• 按顺序处理捕获到的日志事件。

• 转储可以随时进行，跨所有表，针对一个特定的表或者针对一个表的具体主键。

• 以块的形式获取转储，日志与转储事件交错。通过这种方式，日志处理可以与转储处理一起进行。如果进程终止，它可以在最后一个完成的块之后恢复，而不需要从头开始。这还允许在需要时对转储进行调整和暂停。

• 不会获取表级锁，这可以防止影响源数据库上的写流量。

• 支持任何类型的输出，因此，输出可以是流、数据存储甚或是 API。

• 设计充分考虑了高可用性。因此，下游的消费者可以放心，只要源端发生变化，它们就可以收到变化事件。

要求

在之前的一篇博文中，我们讨论了Delta，一个数据充实和同步平台。Delta 的目标是保持多个数据存储的同步，其中一个存储是事实的来源（如 MySQL），其他存储则是派生存储（如 ElasticSearch）。其中一个关键要求是，从事实消息源到目的地的传播延迟要低，并且事件流是高度可用的。无论是同一团队使用多个数据存储库，还是一个团队拥有另一个团队正在使用的数据，这些条件都适用。在我们介绍Delta的博文中，我们还描述了数据同步之外的用例，比如事件处理。

对于数据同步和事件处理用例，除了实时捕获更改的能力外，还需要满足以下要求：

• 捕获全部状态。派生存储（如 ElasticSearch）最终必须存储源的全部状态。我们通过来自源数据库的转储来提供此功能。

• 随时触发修复。我们不将转储视为一次性设置活动，而是打算在任何时候启用转储：跨所有表、在特定表或特定主键上。当数据丢失或损坏时，这对于下游的修复至关重要。

• 针对实时事件提供高可用性。实时更改的传播具有高可用性要求；我们不希望出现事件流停止时间较长（如分钟或更长）的情况。即使在进行修复的过程中，也需要满足这个需求，这样它们就不会停止实时事件。我们希望实时事件和转储事件交错发生，以便两者都取得进展。

• 最小化数据库影响。在连接到数据库时，确保尽可能少地影响数据库的带宽并为应用程序提供读写的能力，这非常重要。出于这个原因，最好避免使用可能阻塞写入流（如表级锁）的 API。此外，还必须加入能够调节日志和转储处理的控制，或者在需要时暂停处理。

• 将事件写到任何输出。对于流技术，Netflix 使用了各种各样的选项，如 Kafka、SQS、Kinesis，甚至 Netflix 特有的流解决方案，如Keystone。尽管将流作为输出可能是一个不错的选择（如当拥有多个消费者的时候），但它并不总是一个理想的选择（如只有一个消费者）。我们希望可以直接写入目的地，而不需要通过流。目标可以是数据存储或外部 API。

• 支持关系型数据库。Netflix 的一些服务使用 RDBMS 类型的数据库，如通过 AWS RDS 使用 MySQL 或 PostgreSQL。我们希望支持将这些系统作为一个源，以便它们能够提供它们的数据供进一步使用。

现有解决方案

我们评估了一系列现有的开源产品，包括Maxwell、SpinalTap、Yelp 的MySQL Streamer和Debezium。现有的解决方案在捕获来自事务日志的实时更改方面类似。例如使用 MySQL 的 binlog 复制协议，或者 PostgreSQL 的复制槽。

在转储处理方面，我们发现现有的解决方案至少存在以下一种限制：

• 在处理转储时停止日志事件处理。如果在转储过程中不处理日志事件，则存在此限制。因此，如果转储量很大，日志事件处理将会在很长一段时间内停止。当下游消费者依赖实时更改的短传播延迟时，这就会是个问题。

• 缺少按需触发转储的能力。大多数解决方案最初都是在引导阶段或在事务日志中检测到数据丢失时执行转储。然而，根据需要触发转储的能力对于引导新消费者（如新的 ElasticSearch 索引）或在数据丢失的情况下进行修复非常重要。

• 通过锁定表来阻塞写流量。一些解决方案使用表级锁来协调转储处理。根据实现和数据库的不同，锁定的时间可以很短，也可以持续整个转储过程。在后一种情况下，写流量被阻塞，直到转储完成。在某些情况下，可以配置专用的读副本，以避免影响主服务器上的写操作。但是，这种策略并不适用于所有数据库。例如，在 PostgreSQL RDS 中，只能从主服务器捕获更改。

• 使用专有数据库的特性。我们发现，有些解决方案使用了高级的数据库特性，这些特性是不能移植到其他系统上的，比如：使用 MySQL 的黑洞引擎，或者从创建 PostgreSQL 复制槽获得一致的转储快照。这阻碍了跨数据库的代码重用。

最后，我们决定实现一种不同的方法来处理转储。它可以：

• 转储事件和日志交错，以便两者都可以进行

• 允许随时触发转储

• 不使用表级锁

• 使用标准的数据库特性

DBLog 框架

DBLog 是一个基于 Java 的框架，能够实时捕获更改并获取转储。转储以块的形式获取，以便可以与实时事件交错，并且不会长时间停止实时事件处理。转储可以通过提供的 API 随时获取。这使得下游用户可以在最初或稍后的修复时捕获完整的数据库状态。

我们设计这个框架是为了尽量减少对数据库的影响。转储可以根据需要暂停和恢复。不管是对于失败后的恢复，还是数据库遇到瓶颈时停止处理，这都很重要。为了不影响应用程序的写操作，我们也不会获取表级锁。

DBLog 允许将捕获的事件写入任何输出，甚至是另一个数据库或 API。我们使用 Zookeeper 来存储与日志和转储处理相关的状态，并将其用于群首选举。我们在构建 DBLog 时考虑到了可插拔性，可以根据需要替换实现（比如用其他东西替换 Zookeeper）。

下面的小节将更详细地说明日志和转储处理。

日志处理

该框架要求数据库针对每个更改的行实时地按提交顺序发出一个事件。事务日志被认为是这些事件的起源。数据库将它们发送到 DBLog 可以使用的传输。我们使用术语“更改日志”来表示该传输。事件的类型可以是创建、更新或删除。对于每个事件，需要提供以下内容：日志序列号、操作时的列状态和操作时应用的模式。

每个更改都序列化为 DBLog 的事件格式，并发送给写入器，以便将其传递到输出。向写入器发送事件是一种非阻塞操作，因为写入器在自己的线程中运行，并在内部缓冲区中收集事件。缓冲的事件按顺序写入输出。该框架允许自定义格式化程序插件，以便将事件序列化为自定义格式。输出是一个简单的接口，允许插入任何需要的目标，例如流、数据存储甚至 API。

转储处理

转储是必需的，因为事务日志的保留时间有限，所以无法用它们重新构造完整的源数据集。转储以块的形式获取，这样它们就可以与日志事件交错，从而允许它们同时进行。为块的每个选定行生成一个事件，并以与日志事件相同的格式进行序列化。这样，如果事件源于日志或转储，下游使用者就无需担心。日志和转储事件都通过同一写入器发送到输出。

可以通过 API 随时调度针对所有表、特定表或表的特定主键的转储。每个表的转储请求按配置好的大小分块执行。此外，可以配置延迟来延缓新块的处理，在此期间只允许日志事件处理。块大小和延迟实现了日志和转储事件处理之间的平衡，并且可以在运行时更新这两个设置。

在选择块时会按主键升序对表进行排序，块中行的主键大于前一个块的最后一个主键。数据库需要有效地执行此查询，这通常适用于实现了主键范围扫描的系统。

图 1 将一个包含 4 列（c1-c4）且以 c1 为主键（pk）的表分块。Pk 列的类型为 integer，块大小为 3。块 2 的选择以 c1 > 4 为条件。

块需要以一种不会长时间阻塞日志事件处理的方式来获取，并且要保留日志更改的历史，这样，如果选取的行是旧值，就不能覆盖来自日志事件的较新的状态。

为了实现这一点，我们在更改日志中创建可识别的水印事件，以便对块选择进行排序。水印是通过源数据库中的一个表实现的。表存储在专用的命名空间中，因此不会与应用程序表发生冲突。存储 UUID 字段的表只包含一行。通过将这一行更新为特定的 UUID 来生成水印。行更新将导致一个最终通过更改日志接收的更改事件。

通过使用水印，转储采取以下步骤：

1. 暂停日志事件处理。

2. 通过更新水印表生成低水印。

3. 为下一个块运行 SELECT 语句，并将结果集存储在内存中，按主键索引。

4. 通过更新水印表生成高水印。

5. 继续将接收到的日志事件发送到输出。监视日志中的高低水印事件。

6. 接收到低水印事件后，开始从结果集中删除所有在低水印之后接收到的日志事件主键的条目。

7. 接收到高水印事件后，将所有剩余的结果集条目发送到输出，然后再处理新的日志事件。

8. 如果出现更多的块，请转到步骤 1。

假定 SELECT 从一致的快照返回状态，该快照表示历史上某个特定点之前提交的更改。或者说：考虑到到那时为止的更改，SELECT 在更改日志的特定位置上执行。数据库通常不公开与 select 语句执行相对应的日志位置（MariaDB 是一个例外）。

我们方法的核心思想是在变更日志上确定一个窗口，它保证包含 SELECT。由于确切的选择位置是未知的，所有与该窗口内的日志事件发生冲突的选中行将被删除。这可以确保选择的块不会覆盖日志更改的历史。通过写入低水印来打开窗口，然后运行选择，最后通过写入高水印来关闭窗口。为了实现这一功能，SELECT 必须读取低水印或之后的最新状态（如果该选择还包括在低水印写之后和读之前提交的写，则没有问题）。

图 2a 和 2b 说明了块选择算法。我们提供了一个示例表，k1 到 k6 为主键。每个更改日志条目表示主键的创建、更新或删除事件。在图 2a 中，我们展示了水印的生成和块的选择（步骤 1 到步骤 4）。在图 2b 中，我们重点看下从位于水印之间的主键结果集中删除选定块的行（步骤 5 到 7）。

图 2a——块选择的水印算法（步骤 1-4）

图 2b——块选择的水印算法（步骤 5-7）

注意，如果一个或多个事务在低水印和高水印之间提交了大量的行更改，则可能会出现大量的日志事件。这就是为什么我们的方法在步骤 2-4 期间会短暂地暂停日志处理，从而保证不会遗漏水印。这样，日志事件处理就可以在以后逐个事件地恢复，最终发现水印，而不需要缓存日志事件条目。日志处理暂停的时间很短，因为步骤 2-4 预计会比较快：水印更新是单个的写操作，而 SELECT 操作有一定的范围。

在第 7 步接收到高水印后，非冲突的块行将被提交写入，以便按顺序发送到输出。这是一个非阻塞操作，因为写入器在单独的线程中运行，允许在步骤 7 之后快速恢复日志处理。然后，日志事件处理将继续处理高水位之后发生的事件。

在图 2c 中，我们使用与图 2a 和 2b 相同的示例来描述整个块选择的写顺序。出现在高水位之前的日志事件首先被写入。然后是块结果的其余行（洋红色）。最后是在高水位之后发生的日志事件。

图 2c——输出写入顺序。日志与转储事件交错。

数据库支持

为了使用 DBLog，数据库需要从提交更改和非过期读取的线性历史中提供更改日志。这些条件由 MySQL、PostgreSQL、MariaDB 等系统来实现，因此框架可以在这些类型的数据库中通用。

到目前为止，我们增加了对 MySQL 和 PostgreSQL 的支持。集成日志事件需要使用不同的库，因为每个数据库都使用了一个专有协议。对于 MySQL，我们使用shyiko/ MySQL -binlog-connector来实现 binlog 复制协议，以便从 MySQL 主机接收事件。对于 PostgreSQL，我们通过wal2json插件使用复制槽。更改通过由 PostgreSQL jdbc驱动程序实现的流复制协议接收。对于捕获的每个更改的模式，MySQL 和 PostgreSQL 的确定方式是不同的。在 PostgreSQL 中，wal2json 包含列名和类型以及列值。MySQL 模式的更改则必须通过接收的 binlog 事件进行跟踪。

转储处理是使用 SQL 和 JDBC 集成的，只需要实现块选择和水印更新。MySQL 和 PostgreSQL 使用相同的代码，其他类似的数据库也可以使用相同的代码。转储处理本身不依赖于 SQL 或 JDBC，并且允许集成满足 DBLog 框架要求的数据库，即使它们使用不同的标准。

图 3——DBLog 高阶架构

高可用性

DBLog 使用主动-被动架构。一个实例是主动的，其他的是被动的。我们利用Zookeeper进行群首选举，从而确定活动实例。领导权是一种租约，如果没有及时更新，就会丢失，让另一个实例接管。我们目前为每个 AZ 部署一个实例（通常我们有 3 个 AZ），因此，如果一个 AZ 宕机，另一个 AZ 中的实例可以继续处理，保证总停机时间最少。被动实例也可以跨区域，但建议在与数据库主机相同的区域内进行操作，以保持较低的更改捕获延迟。

生产使用情况

DBLog 是 Netflix MySQL 和 PostgreSQL 连接器的基础，这些连接器在Delta中使用。Delta 自 2018 年起用于生产，用于 Netflix studio 应用程序中的数据存储同步和事件处理。在 DBLog 之上，Delta 连接器使用自定义的事件序列化器，因此，在将事件写入输出时会使用 Delta 的事件格式。Netflix 特有的流被用作输出，比如Keystone。

图 4——Delta 连接器

除了 Delta 之外，DBLog 还用于为其他 Netflix 数据移动平台构建连接器，这些平台有自己的数据格式。

敬请关注

DBLog 还有一些本文没有涉及的功能，比如：

• 能够在不使用锁的情况下捕获表模式。

• 模式存储集成。存储发送到输出的每个事件的模式，并在每个事件的有效负载中包含到模式存储的引用。

• 单调写模式。确保一旦针对特定行写入了状态，之后就不能写入稍早的状态。通过这种方式，下游消费者只会看到前向状态转换，而不需要在时间上来回切换。

我们计划在 2020 年开源 DBLog 及更多文档。

原文链接：

DBLog: A Generic Change-Data-Capture Framework