Flink CDC 在阿里云实时计算Flink版的云上实践

摘要：本文整理自阿里云高级开发工程师，Apache Flink Committer 阮航老师在 Flink Forward Asia 2024 生产实践（三）专场中的分享，主要分为以下四个方面：

Flink CDC & 实时计算 Flink
CDC YAML 核心功能
CDC YAML 典型应用场景
Demo & 未来规划

Flink CDC & 实时计算 Flink

1.1 Flink CDC 简介

Flink CDC 在经过多个版本的发布后，已经不仅仅是用于处理 MySQL 或其他数据源的 CDC Source。如今，Flink CDC 是一款用于处理流批数据的分布式数据集成工具，支持通过 YAML 格式描述数据传递过程中的转换和路由等数据集成工作。

1.2 实时计算 Flink 版集成 Flink CDC

在最新版本的实时计算 Flink 版中，我们已经集成了开源 Flink CDC 的全部功能。我们专注于提供数据开发中的数据摄入部分，支持用户通过 YAML 配置文件进行完整的数据集成作业开发。采用 YAML 配置的方式降低了使用门槛，同时，我们还预设了一些常用的模板供用户选择。

在数据输入环节，我们支持多种常见的数据入湖入仓链路，例如从 MySQL 到 Paimon、Hologres、StarRocks 等同步链路。这些同步链路所依赖的连接器均由系统自动识别并添加到作业中，无需用户自行管理。

此外，数据摄入功能还会帮助用户管理整个 CDC YAML 作业的生命周期，包括启动、停止以及从 savepoint 重启等操作。同时，我们也提供了版本管理的便捷功能，如版本回滚等。为了满足用户对监控的需求，我们还提供了丰富的数据同步作业监控指标。

下方提供了数据摄入 CDC YAML 的文档链接，感兴趣的朋友可以点击了解详情。（https://help.aliyun.com/zh/flink/user-guide/develop-a-yaml-draft?spm=a2c4g.11186623.help-menu-45029.d_2_2_3.5cf92046K7da4U）

CDC YAML 核心功能

第二个部分介绍目前 CDC YAML 已经覆盖到的一些核心的功能。

2.1 支持的同步链路

目前，CDC YAML 数据摄入功能支持的同步链路 Source 端除了包括广泛使用的 MySQL外，我们还支持 Kafka 链路，该链路主要同步的是原始的 Binlog 数据。用户可以将 Binlog 数据同步到 Kafka 中，然后再通过 Kafka 同步到其他目标端。同时，CDC YAML 还支持 Schema Evolution（模式演进）以及 Schema Sync（模式同步）变更，具备整库同步的能力，能够将数据变更同步到下游系统。

目前，下游目标端主要包括 Paimon、StarRocks、Hologres 和 Kafka。对于 Kafka，我们目前支持两种同步方式：一种是同步到事实表，Flink Upset Kafka 连接器可以直接读取这些数据；另一种是同步到 Kafka 并存储其原始的 Binlog 日志，供用户进行后续处理。

2.2 Transform 和 Route

CDC YAML 还具备一些额外的功能，即 Transform 和 Route。Transform 模块旨在数据同步过程中进行额外的计算和处理。例如，可以添加计算列，根据数据列计算出额外的列，或者在读取过程中添加元数据列，如操作类型、表名或数据库名等，并将这些信息一并传输到下游。在计算过程中，我们提供了丰富的内置函数，并且支持用户自定义 UDF（用户定义函数）。

为了方便用户进行作业的迁移，我们对 SQL 作业中的 Flink ScalarFunction 进行了兼容，用户可以无缝地在 Transform 部分使用自己的 ScalarFunction。此外，Transform 模块还支持重新指定下游主键或分区键，这主要是为了处理一些下游表在写入时可能存在的额外依赖。同时，我们也支持对数据进行过滤，清理不需要的数据。

在 Route 模块中，我们主要关注的是上下游的映射关系配置。常见的整库同步场景中，可以将上游的一张表一对一同步到下游的一张表。在分库分表场景中，可能涉及多个库的多张表，我们希望将它们同步到目标端的同一张表中，可以通过 Route 进行配置。此外，Route 还支持批量定义下游的 Sink 表名，例如为所有表名统一添加版本前缀等。

2.3 数据同步指标

此外，在数据摄入部分，我们还额外提供了一系列丰富的指标，以帮助用户全面了解数据同步的进度及相关信息。CDC 支持了全增量一体化同步，在全量同步阶段，CDC 会将每个表切分成不同的分片进行处理，因此，了解当前处理到哪些分片对于监控同步进度至关重要。

我们提供了指标用于判断当前是处于全量同步阶段还是增量同步阶段。同时，我们还根据表提供了全量阶段已处理分片数量、未处理分片数量、已处理表数量以及未处理表数量等具体信息，以便用户清晰地查看同步进度。

在数据相关部分，我们提供了一些关键数据指标。例如，最新读取到的数据时间戳，以及常见的延迟指标（Lag）。此外，指标包含读到的数据条数，分为全量同步阶段读取到的数据条数和整个作业启动后读取到的数据条数，并提供了表级别的数据条数监控。

2.4 其它功能

此外，我们还提供了一些更为丰富的功能。例如，细粒度 Schema 变更策略。在数据同步过程中，为了避免在下游执行一些潜在危险的操作，如 Drop Table 的变更，我们提供了细粒度的表变更序列控制，允许用户进行过滤和跳过。关于这部分的详细内容，我们将在后续进行更深入的介绍。

另外，我们还支持变更的宽容模式，支持更多类型的变更操作，如 Drop table、 Truncate table 等。此外，我们还支持原始的 Binlog 数据同步，以满足用户多样化的需求。

2.5 YAML 与 SQL 比较

相较于 SQL 作业开发，我们对 CDC YAML 开发方式进行了简单的对比分析。传统的 SQL 作业开发无法支持 Schema 变更，需要事先定义 Schema，且 Schema 在作业启动后无法变更，这限制了其灵活性。此外，由于 SQL 依赖于 RowData 传递数据，这会破坏 Binlog 的原始格式。例如，将一条 Update 消息拆分为 Update Before 和 Update After 两条消息，这导致 SQL 作业无法支持同步原始 Binlog 的场景。另外，SQL 方式通常仅支持单表的读写操作。

而 CDC YAML 为数据同步场景设计，支持自动同步 Schema 变更，并提供了细粒度 Schema 变更控制。更重要的是，CDC YAML 在传递内容时保留了原始的 Changelog 格式，包括 Data Change Event 和 Schema Change Event。此外，CDC YAML 还支持多表的读写操作。

2.6 YAML 与 CTAS/CDAS 比较

对于熟悉实时计算 Flink 的用户而言，SQL 提供的 CTAS 和 CDAS 两种语法是支持数据同步的工具。然而，由于 CTAS/CDAS 是基于 SQL API 的，依赖于 RowData 格式传递数据，无法完成同步清理表等操作，而且 Schema 变更同步需要有新数据到来时才会同步。CTAS/CDAS 会破坏原有的 Changelog 结构，使得无法进行原始的 Changelog 同步。

而在 CDC YAML 中，我们支持了更多类型的 Schema 变更，如 Truncate Table 和 Drop Table 等操作，支持原始 Binlog 同步，提供了细粒度的 Schema 变更策略。通过 Route 模块，可以灵活地定义目标表的名称。此外，CDC YAML 还支持数据过滤功能，以及通过 Transform 模块裁剪字段，省略不需要的数据。

2.7 YAML 与 DataStream 比较

这里我们简要对比一下 YAML 与编写 DataStream 作业的优势。DataStream 作业的主要挑战在于，用户需要深入了解 Java 和分布式系统，包括 Flink 框架及其相关技术。此外，用户还需要自行管理依赖、熟悉相关工具，并且 DataStream 作业的代码往往难以复用。

相比之下，CDC YAML 在多个方面展现出显著优势。它特别适用于包括初级用户在内的各类用户，因为 CDC YAML 隐藏了底层细节，使得开发过程更加简洁。采用 YAML 格式，不仅易于开发，而且便于理解。更重要的是，CDC YAML 作业的整体配置清晰明了，一目了然，极大地促进了配置的复用性。

CDC YAML 典型应用场景

第三部分，主要介绍 CDC YAML 实际上可以帮助处理怎样的问题，它典型的应用场景是怎样的。

3.1 整库同步

最简单的应用场景是整库同步。例如，在 MySQL 中存在 Products、Customers 和 Orders 三张表，我们需要将这些表同步到 Paimon 中，以便后续进行高效的查询和分析。为实现这一功能，我们只需简单地采用如图的 YAML 作业进行同步，只需定义好 Source 端 MySQL 的相关内容，明确需要同步的表，然后配置 Sink 端的连接信息，即可完成整库同步的工作。

在展示了最简单的整库同步之后，我们来看看更复杂的同步场景。在某些场景下，我们可能需要定义 Transform 和 Route 规则，以捕获并添加额外的信息到下一步的表中。

举个例子，对于 Customers 表，我们可以添加一个名为 db 的列，将 Customers 表所在的数据库名（app_db）写到下游。此外，我们还可以利用内置函数对字段进行处理，比如将 name 字段转换为大写。同时，我们还可以自定义目标表的表名，此处通过 Route 模块定义 Source 和 Sink 之间的映射关系并写入数据。

在整库同步过程中，我们实现了一体化的全增量数据同步。蓝色表示增量数据，灰色表示全量数据。例如，Products 表会映射到 Products_v0 表，Orders 表会同步到 Orders_v0 表，而 Customers 表则会同步到 Customers_v1 表。

完成全量数据同步后，系统会自动切换到增量同步模式。如果 Customers 表中增加了新的列，如 age 列，系统也会自动进行同步。

在整库同步过程中可能会遇到下游不支持某些变更类型的情况，在同步到 Hologres 链路上，我们额外提供宽容模式来解决这一问题。

宽容模式的主要目的是减少 Schema 变更的频率，因为在 CDC 同步数据时，Schema 变更需要 Sink 端将已有的数据整体刷新到下游存储中，这是一个较重的操作。因此，我们希望通过宽容模式，将部分 MySQL 类型映射到下游的统一类型中，从而在类型变化时避免进行 Schema 变更，直接跳过这些变更，使数据同步更加高效。

目前，宽容模式仅支持 Hologres 写入端。用户可以在 Hologres Sink 模块中，设置 sink.type-normalize-strategy 选择对应的宽容模式策略。以图中策略为例，将 MySQL 的类型仅映射到 Hologres 的 BIGINT 和 TEXT 两个类型中，其他策略的具体的映射关系可以在文档中查看。在同步过程中，数据变更会正常同步。但是，如果遇到 Schema 变更，比如调整 VARCHAR 类型的列长度，由于这两个类型在下游 Hologres 中都映射为 TEXT 类型，因此可以跳过 Schema 变更的处理，从而提高数据同步的效率和稳定性。

接下来，我们要讨论的是整库同步到 Kafka 的功能。实际应用中，当多个作业依赖于 MySQL 的同一张表时，频繁地建立数据库连接可能会导致数据库压力过大。整库同步后的 Upset Kafka 表可以被当作原来的 MySQL 表来读取，从而减轻系统压力。

下面是整库同步到 Kafka 作业的简单展示。首先，我们需要定义 MySQL 的数据源。然后，在 Sink 端简单地配置连接信息，就完成了整个作业。在整库同步到 Kafka 时，可以通过 Route 模块定义下游的 Topic 名称。如果没有特别定义，默认 Topic 名称会使用数据库名加上表名的组合。

3.2 Binlog原始数据同步Kafka

除了将整个表的数据完全同步到 Upset Kafka 之外，还有一种场景是用户可能需要完整的表变更信息，即 Binlog 的原始内容。这个功能在 CTAS/CDAS 中无法支持，但我们在 CDC YAML 中进行了相应的支持。

如果用户需要将 Binlog 数据原始地同步到 Kafka，可以按照以下方式进行配置。首先，简单地定义 MySQL 的连接信息。然后，如果用户需要 op_ts（操作时间戳）这一元数据，可以额外添加一个 op_ts 的 Metadata 列，系统会在对应的 Topic 中自动添加这一列。

目前，我们支持了 Debezium json 和 canal json 两种变更格式。默认使用 debezium json 类型，由于 Update 操作的格式比较复杂，所以图中展示的是 Update 的格式，它包括 Before 和 After 两个内容，以及操作类型和来源信息。

3.3 分库分表同步

下一种常见的同步场景是分库分表同步。数据库中一张表可能会包含过多的数据，用户会根据需要将其拆分为多个表进行存储。然而，在下游的分析过程中，需要将这些分散在多个表中的数据合并为同一张表，以便进行整体的分析查询。

为了实现这样的功能，我们需要在 Route 模块中设置多对一的表映射关系。例如，我将 app_db 下所有以 customers 开头的三张表都写入到 Paimon 的 Customers 这一张表中。因此，在整个同步过程中，这三张表的数据都会同步到这个同一张表中，从而完成分库分表同步的功能。

3.4 细粒度 Schema 变更策略

在数据同步过程中，MySQL 可能会发生一系列变更操作。、主要的变更操作包括新增表、新增列、修改列、删除表、删除列或清空表等。其中如清空表操作是比较危险的，因为如果直接同步到下游，会导致下游数据被完全清空，用户可能不希望进行这些变更的同步。

为了解决用户在同步过程中需要的多种变更场景，我们提供了细粒度Schema变更策略。通过图示，细粒度Schema变更策略在容错性和演进性上进行了区分。

IGNORE 模式会完全忽略全部变更操作，不进行同步。EVOLVE 模式会在同步过程中完整地将上游的变更直接同步到下游，而 EXCEPTION 模式则不允许任何形式的变更，一旦遇到 Schema 变更就会报错。

为了平衡容错性和数据安全性，我们比较推荐使用 LENIENT 和 TRY_EVOLVE 这两种模式。LENIENT 会以兼容的方式进行变更，TRY_EVOLVE允许目标端跳过一些不支持的变更。LENIENT 变更模式可以兼容已有的作业，确保使用同步后的表的作业能够继续正常运行，同时让数据更加安全。

在 LENIENT 模式下，它不会同步一些危险的操作，如清空表、删除表等，并且会避免一些可能导致下游消费数据不兼容的变更操作。例如，在同步修改列名的操作时，LENIENT 模式不会直接将列名修改，因为这可能导致下游的消费作业无法兼容。它会保留原列，并以添加新列的形式进行兼容。

TRY_EVOLVE 模式则允许 Sink 在尝试同步时，对其不支持的变更行为显示异常，并支持跳过 Schema 变更。

此外，为了更灵活地进行变更控制，我们还支持在 Sink 模块中使用 Include 或 Exclude schema change 等操作来控制某一类的变更操作。在这种模式下，你可以根据需要对特定的变更进行额外的配置。

3.5 自动捕获新增表

下一个场景是新增表，实际上在我们作业启动之后，用户很可能需要变化捕获的表。例如作业同步的表是 V0，但是在作业运行之后，需要将其它表，例如 V1 和 V2 再进行同步。对于数据同步的开发者来说，当面临既要同步已存在的表的历史数据，又要同步新产生的数据时，应该如何操作呢？

此外，还有一种场景是，在同步作业启动时，已经能够正常同步具有特定前缀（如Customers）的表，例如V0、V1、V2。然而，在作业运行过程中，上游可能会创建一个符合该前缀的新表（如V3），且这张新表没有历史数据。针对这种新表，我们应该如何确保其能够被正确同步呢？

实际上，我们已经支持了上述两种场景，但它们的支持方式有所不同，需要大家注意。

对于第一种场景，在需要同步已有表的历史数据时，我们需要对 Source 开启scan.newly-added-table.enable配置，并从 Savepoint 重启重新捕获那些需要同步历史数据的表。

对于第二种场景，即在作业已经启动并匹配到某些表后，上游又新创建了符合匹配条件的表，且不需要同步这些新表的历史数据时，我们需要开启scan.binlog.newly-added-table.enable 配置。这样，作业无需重启，系统会自动将新匹配的表进行创建，并开始同步新创建表的变更数据。

3.6 指定位点 + 全量重刷

在作业同步过程中，如果因版本兼容问题导致解析 Binlog 时遇到错误，且无法从 Savepoint 停止的位点继续恢复启动，以往可能需要完全无状态地重启作业，进行全增量同步。在支持指定位点启动后，我们可以在启动作业时指定到停止作业的位点，从而跳过那些无法恢复的错误。通过这种方式，我们可以选择重刷数据以修正错误，而无需进行全量同步。

3.7 MySQL CDC 企业级性能优化

在 MySQL CDC（Change Data Capture）数据摄入过程中，我们经常收到关于Binlog 消费速度过慢的反馈。为此，我们进行了企业级性能优化，主要包括以下四个方面：

首先，在 Binlog Bump 时，我们对 Debezium 的参数配置进行了改进。通过优化数据拉取间隔和缓存队列长度等相关参数，我们实现了性能提升。经过测试，与开源版的 Flink CDC 相比，这种优化可以带来约 11% 的速度提升。

其次，我们实施了过滤无关表数据的策略。由于 MySQL Source 通常消费整个数据库的数据，而实际作业可能只需要其中几张表的数据，因此过滤掉无关表的数据可以加速数据读取。这一优化的效果主要取决于过滤的数据量。

第三，我们实现了并行解析 Binlog 的操作。与开源版本仅支持单线程解析不同，我们在企业级 MySQL CDC 中将其优化为多线程解析。这一改变使得解析效率得到了显著提升，与单线程相比，效率提升了约 14%。

最后，我们进行了并行序列化的优化。序列化是一个耗时较多的过程，通过火焰图分析，我们发现 Event 到 SourceRecord 以及 SourceRecord 到 JSON 的序列化过程耗时尤为显著。因此，我们将序列化优化为并行序列化并保持顺序，这一优化使得解析速度相比开源版本提升了 42%。

综合以上四个优化策略，如果数据库中只包含一张表的数据，那么效率大约可以提升 80% 左右。多数情况下，数据库中会存在很多业务表，一个Flink作业通常只需要其中部分表的数据，那么过滤无关表的优化能够获得更大的收益，Source 的读取甚至能有十倍的性能提升。

Demo & 未来规划

最后一部分，简单展示一下数据摄入功能Demo以及未来规划。

4.1 Demo：整库同步Paimon

整库同步Paimon

第一个 Demo ，主要展示 MySQL 整库同步 Paimon 链路，这里展示的数据库主要包含 Customers、Products 和 Shipments 三张表，并展示了具体的作业指标。在作业启动后，在数据库进行一些变更后，再在 Paimon 查看变更同步结果。

之为了观测到删除列的变更，在作业中配置了 Evolve 变更策略。作业开发时，敏感信息可以在变量管理部分以密文形式保存，作业部署并选择无状态启动将作业跑起来。

作业正常运行后，视频中展示了监控指标内容，主要关注数据摄入部分。IsBinlogReading 的指标为1表示已进入增量同步阶段，同步了三个数据表，整库同步的数据量也是九条。

在创建 Paimon 的 Catalog 中，可以看到对应的表已经成功建立。之后利用ETL的调试功能，在 Catalog 中选择了具体的数据进行查看。

随后，在 Customers 表中进行数据变更，以验证是否能正常同步。这里变更包括插入两条新数据，添加 Age 列，更新数据并删除一个列。提交修改后，我重新选择了 Customer 表以查看结果。Paimon 已正常写入更新的数据，并且数据也已同步。

4.2 Demo：同步Binlog到Kafka

同步Binlog到Kafka

接下来，我将展示整库同步 Binlog 到 Kafka。同样，这里选择了三张表，共九条数据进行同步，通过 route 指定 Topic的名字。

部署整库同步作业后，作业正常启动。检查了监控告警，发现已经完成了从全量到增量的同步切换。在全量阶段，每张表有三条数据，因此共九条数据已同步完成。

来到 Kafka 这边，可以看到三个 Topic 已成功建立。我查看了具体的表列数据，选择了更早的时间点进行查询，这里使用的是默认的 debezium json 格式来保留数据。在 Customer 表中进行了 Schema 变更和修改，包括插入新数据、删除和更新数据。在 Kafka 中查询时，可以看到数据已正常同步。

4.1 未来规划

最后，简单介绍一下未来规划。首先，我们计划添加脏数据处理功能，以完善整个数据同步链路。这包括收集并展示那些被跳过或错误处理的数据，以便客户能够清晰地了解数据同步过程中的情况。

其次，针对数据同步可能对上游产生的压力问题，计划通过数据限流功能解决，使得用户能够自主控制数据的读取速度，从而更好地管理数据同步过程对上游资源的影响。

此外，我们还计划拓宽整体的数据同步链路，以支持更多的数据源和目标端。在此，我也简单展示了一下数据摄入 Beta 版链接。下方有基于 Flink CDC 打造企业级实时数据同步方案，如果大家对此感兴趣，可以自行尝试并体验我们的新方案。

文档链接：

数据摄入 Beta 版：https://help.aliyun.com/zh/flink/user-guide/develop-a-yaml-draft?spm=a2c4g.11186623.help-menu-45029.d_2_2_3.53b92058TtBz1e&scm=20140722.H_2846225._.OR_help-V_1
开源 Flink CDC：https://nightlies.apache.org/flink/flink-cdc-docs-stable/

如何通过 Flink CDC 实现高效、可靠的实时数据同步方案？

复制下方链接或点击阅读原文

即刻了解详细方案：

https://www.aliyun.com/solution/tech-solution/flink-cdc-realize-data-synchronization?utm_content=g_1000401146