Flink CDC 3.3 版本解读

摘要：广义上，能够捕获数据变更的技术都可以称为 CDC (Change Data Capture)。通常我们说的 CDC 技术主要针对数据库的变更，是一种用于捕获数据库中数据变更的技术。

导读本次分享题目为 Flink CDC 3.3 版本解读。

主要介绍：

1. Flink CDC 简介

2. Flink 3.3 版本概览

3. Flink 3.3 版本核心特性和改进

4. 未来规划

5. 问答环节

分享嘉宾｜阮航阿里云阿里云技术专家\Apache Flink Committer

编辑整理｜张静瑜

内容校对｜李瑶

出品社区｜DataFun

广义上，能够捕获数据变更的技术都可以称为 CDC (Change Data Capture)。通常我们说的 CDC 技术主要针对数据库的变更，是一种用于捕获数据库中数据变更的技术。

具体应用场景包括：

数据分发：将一个数据源的数据同步到多个下游。数据集成：面向数据仓库/数据湖进的 ETL 数据集成。数据迁移：用于数据备份、系统容灾，确保数据安全性和业务连续性。

2. Flink CDC 概述

Flink CDC 是 Flink 深度结合数据库的 CDC 技术，打造的全增量一体化读取的流式数据集成框架。它是实现 Streaming Lakehouse 的关键技术。通过增量快照框架帮助用户完成全增量一体化的数据同步，提供无锁读取、断点续传和并发读取能力。

Flink 3.3 版本概览

Flink 3.3 版本是一个重要的更新版本，主要针对连接器（Connector）和转换模块（Transform）进行了多项改进和优化。该版本共处理了 107 个 issue，合并了 164 个 Commits，共有 37 位贡献者参与了开发工作。

1. Flink CDC 3.3 概览

连接器（Connectors）：新增支持：新增 OceanBase 和 MaxCompute pipeline 连接器的支持，覆盖了更多下游系统。MySQL 连接器：支持了 TINYINT 类型映射指定，并支持解析无锁变更的 DDL。Paimon 连接器：升级底层 Paimon 依赖，完善 DDL 事件同步。PostgreSQL 连接器：支持了 OP_TYPE 的元数据列，并优化了初次快照的耗时。增量快照框架：异步分片：在全量阶段，对表的分片改为异步分片方式。Transform 模块：新增内置函数。支持逻辑删除。版本兼容性：Flink 兼容版本：Flink CDC 3.3 版本开始，兼容 Flink 1.19 和 1.20 两个版本。

2. Flink CDC 社区

截至目前，Flink CDC 在 GitHub 上已获得 5.9k Stars 和 2k Forks，吸引了 143 位贡献者加入开发。社区钉钉群成员数达到了1万以上，形成了一个活跃的技术交流平台。

Flink 3.3 版本核心特性和改进

Flink CDC 3.3 版本在连接器、转换模块（Transform）、增量快照框架等方面进行了显著的优化和功能增强，进一步提升了系统的稳定性和功能性。

1. 新增 OceanBase & MaxCompute pipeline 连接器

OceanBase 连接器：OceanBase 是一款原生分布式关系型数据库，具备高性能和高可用性，能够支持海量数据和高并发场景。现在可以作为 YAML pipeline sink 使用。MaxCompute 连接器：MaxCompute 是阿里云提供的分布式大数据处理平台，广泛用于数据分析、报表生成和机器学习等场景。现在可以作为 YAML pipeline sink 使用。CDC YAML 链路覆盖源端支持：Flink CDC 支持从 MySQL 系列的数据库（如 MySQL、PolarDB）作为源端，通过 CDC YAML API 提供的 Schema 变更同步和整库同步功能，完成数据的实时同步。目标端支持：目标端支持多种数据湖仓，包括 Paimon、StarRocks、Elasticsearch、Kafka、OceanBase 和 MaxCompute。后续版本将继续扩展对上下游生态的支持。

2. Transform 模块增强

AI model 集成：Flink CDC 3.3 版本支持用户通过内置的 AI 模型对捕获的原始数据进行丰富化处理，如数据分类、实体识别和特征提取等。用户可以通过配置 AI 模型，在 Transform 模块中调用这些功能，生成额外的计算列。

示例配置

模型注册：在 pipeline 配置中注册需要的 AI model。目前内置 OpenAI Chat 模型及 Embedding 向量化模型。模型调用：在 Transform 模块的 projection 配置中调用这些模型，生成额外的计算列。例如，添加一个使用chat模型生成的分类标签列和一个使用 get_embedding 模型生成的特征向量列。
添加 OP_TS 元数据：OP_TS 是数据变更事件发生的时间戳，对于数据审计和追溯至关重要。此前，Transform 模块仅支持 namespace、schema、table_name和 data_event_type 四种元数据，而用户在实际场景中常需要 OP_TS 信息。

示例配置

用户可以在 MySQL 数据源的配置中添加 metadata.list 配置项，指定需要传递的元数据字段，例如 op_ts。在 Transform 模块中，用户可以像使用其他元数据字段一样使用 op_ts。

工作原理

框架层面的修改在Flink CDC 3.3 中，新增了一个 supportedMetaDataColumns 类，用于封装元数据列的名称、类型及解析方式。在 DataSource接口中增加了 supportedMetaDataColumns方法，允许 source 返回其支持的所有额外元数据列。Transform 模块的元数据处理在 Transform 模块中，系统首先尝试匹配固定的元数据列（如 namespace、schema 等），若未匹配到，则尝试匹配源端支持的额外元数据列（如 OP_TS）。若仍未匹配到，则最终匹配表的物理列。用户可以在 Transform 模块中像使用其他元数据一样使用 OP_TS。例如，将 OP_TS 与其他元数据组合生成唯一标识符，或将其作为数据审计和追溯的依据。连接器的修改连接器需要实现 DataSource 接口中新增的 supportedMetaDataColumns 方法，定义支持的额外元数据列（如 OP_TS）。在解析数据变更事件时，连接器需要将 OP_TS 元数据嵌入到 DataChangeEvent 的 Meta映射中，以便在 Transform 模块中使用。

支持逻辑删除：逻辑删除（软删除）指标记数据为已删除而不实际移除，利于审计和数据恢复。

配置与实现

配置逻辑删除：在 Flink CDC 的 Transform 模块中，用户可以通过配置 converter-after-transform 来启用逻辑删除功能。这一过程需要结合 data event type 使用，以标记数据是否已被删除。例如，在从 MySQL 表同步到 Paimon 表的过程中，可以在建表时将 data event type 添加到表结构中，并可能重命名为 OP type 以便于识别。处理变更事件：当一个表发生变更事件时（如插入、更新或删除操作），Transform 模块会根据配置对这些事件进行处理。假设有一个名为 customers 的 MySQL 表，首先插入了两条记录（ID 分别为 1 和 2）。随后，这两条记录被标记为删除。在Sink 端接收到的变更顺序将是：首先增加两条记录（ID 1 和 2），对应的 op_type 值为 +I 表示插入。对于删除操作，逻辑删除机制会将这些删除操作转换为插入操作，并设置 op_type 为 -D 表示删除。如果涉及更新操作，则 op_type 将被更新为 +U。

增加内置函数支持：新增操作时间戳的 TimestampAD、TIMESTAMPDIFF、UNIX_TIMESTAMP 等内置函数。自动类型转换：在 Transform 模块中定义主键时，系统会自动将字段类型转换为非空（NOT NULL）类型。这是因为在许多情况下，下游的目标端（如数据库或数据仓库）要求输入数据不能包含空值。问题修复与优化：TO_TIMESTAMP 比较修复：修复了 TO_TIMESTAMP 在 Transform 模块中比较时可能触发的报错问题。分库分表场景优化：修复了分库分表场景下因 schema 不同导致的 Transform 模块报错问题，支持统一的 ETL 处理。默认列选择：当用户未提供 projection 映射时，Transform 模块默认选择所有列，简化了用户配置并避免了验证错误。精度丢失修复：修复了 Decimal 类型转换后精度丢失的问题。

3. Flink CDC 增量快照框架

下面介绍增量快照框架方面的一些改进。首先简单介绍一下 Flink CDC 的增量快照框架。Flink CDC 通过增量快照框架可以帮助用户全增量的一体化同步数据，并且提供了无锁变更、断点续传和并发读取的能力。在全量阶段读取的是用户的历史数据，会把需要捕获的表切分为多个分片，分发给对应的 source reader 进行并发的读取。在全量阶段数据读取完后，通过增量快照框架会全增量自动切换为增量读取部分，通过 MySQL 的 Binlog 去读取数据，这种增量读取是单并发的，从而完成整体的全增量一体化读取。目前， SQL Server、 Postgres 等都接入到了增量快照框架中，支持了无锁变更、断点序传、并发读取的能力。

Flink 3.3 版本对增量快照框架进行了优化，特别是在表切分和数据读取效率方面，进一步提升了数据同步的性能和稳定性。

Flink CDC 增量快照框架异步分片

在全量阶段，Flink CDC 需要将需要捕获的表切分为多个分片（chunk），以便并发读取。然而，在某些情况下，由于切片列不满足均匀切分的条件，需要执行 SQL 查询来获取数据的最大值和最小值进行切分。这种操作可能会导致与数据库的长时间交互，以致延迟整个表的切分过程，造成某段时间内没有数据分片下发到下游的情况。

为了缓解这一问题，Flink CDC 3.3 版本引入了一个独立的 snapshotSplit 线程来异步执行表的切分操作。具体来说：

异步切分：通过 chunk splitter 异步地对表进行分片。队列管理：切分后的数据分片会被存入 remainingSplits 队列中。高效读取：snapshot-splitting thread 会从队列中获取并处理这些分片，提高了数据读取的速度。

这种优化确保了即使在表切分过程中，下游也能持续接收到数据分片，避免了因切分延迟而导致的读取中断问题。

4. 连接器优化和改进

MySQL 连接器改进自动读取新增表：通过 scan.binlog.newly-added-table.enabled 配置，支持自动读取已匹配表的变更数据，无需重启作业。无锁 schema 变更工具支持：支持解析 gh-ost 和 pt-osc 工具产生的 DDL 变更事件，确保 schema 变更时不影响业务。字段类型映射优化：支持将 Tinyint(1) 映射为布尔类型或保留为整数类型。表注释和行注释同步：支持同步表的注释和行的注释到目标端。JSON 数据格式一致性：修复全量和增量阶段 JSON 数据格式不一致的问题，支持通过配置统一格式。问题修复：修复了新增表处理时的死锁问题、gtid 未开启时的重复读取问题，以及 MySQL 8.0.17 和 8.0.18 版本的 Binlog 解析兼容性等问题。
Paimon 连接器改进支持同步列的默认值。支持处理 Truncate table 和 Drop table 变更。Paimon 依赖版本升级至 0.9.0。修复了 Paimon Sink 重复提交导致的作业失败问题。Postgres 连接器改进减少了不必要的 schema 查询，优化了初次启动时间。支持发送 Heartbeat 心跳包，及时更新 LSN，避免磁盘存储过多无用数据。新增 op_type 元数据列。Kafka 连接器改进支持 Sink 配置 topic 的映射关系，保留原始 tableid 信息。Flink CDC Base 改进全量阶段异步分片。支持更多指标（如全量阶段数据），提供表级别细粒度指标。

5. 支持 Flink 1.19+

Flink CDC 3.3 版本仅支持 Flink 1.19 和 1.20 系列版本。在历史版本中，Flink CDC 曾兼容多个 Flink 版本，例如 3.1.1 和 3.2.1 版本均支持四个 Flink 版本。然而，这种多版本兼容策略导致部分新功能和新接口无法正常使用。因此，在 Flink CDC 3.3 版本中，参考外部连接器的维护方案，仅支持最新的两个 Flink 版本，即 1.19 和 1.20 系列版本，放弃对 1.19 以下版本的支持。

6. 其他改进

Pipeline 模块的配置已改为可选，用户无需强制配置，降低了使用复杂度。提供了 CDC UP 工具，帮助用户快速构建和测试 Flink CDC 作业。对 CDC YAML 中各个组件的资源清理进行了优化，例如连接池和线程池等资源在停止时会进行合理关闭。在 CDC CLI 工具中，支持通过命令行配置覆盖 Flink conf 中的配置。分库分表场景优化:在分库分表 Route 合并时，支持 Timestamp 类型的合并。修改了 bigint 和 double 类型合并的行为，统一合并为 double 类型，而不是像 3.3 版本以前合并为 string 类型。修复了 Array 类型和 Map 类型可空属性的设置问题。支持处理 int64 长度的整数字面量，而此前仅支持 32 位整数字面量。修复了 Flink cdc pipeline 作业无法从 Flink 1.20 状态恢复的问题。支持 Array 和 Map 的复杂嵌套类型。

未来规划

1. 丰富上下游生态

在未来版本中，Flink CDC 将继续支持更多的上下游数据系统，例如 Iceberg / JDBC pipeline sink 等，以拓宽用户的使用场景，为用户提供更丰富的上下游适配能力。

2. 升级 Flink 到 1.20

Flink CDC 计划将其依赖的 Flink 版本升级到 1.20，在确保兼容性的前提下，利用高版本 Flink 提供的新功能和性能优化。

3. 支持 Batch Mode 运行

为了更好地支持全量数据快照同步，Flink CDC 将增加对批处理模式的支持。这一改进使得用户可以方便地对历史数据进行订正和补回，特别适用于需要定期校正或补充数据的场景。

4. 脏数据处理

Flink CDC 计划支持脏数据处理和收集功能，以完善整体的数据同步链路。通过这一功能，用户可以更有效地管理和处理同步过程中出现的异常数据，从而提升整体使用体验。

5. 数据限流

为了降低对上游数据库的压力，避免影响正常的业务操作，Flink CDC 将引入数据限流功能。该功能允许用户控制数据同步的速度，确保在全量和增量阶段都能平稳运行，不影响其他业务操作。

6. 更多变更类型支持

Flink CDC 将考虑支持更多的 MySQL 变更类型，如 RENAME TABLE 和 ALTER TABLE COMMENT 等。通过这些新增功能，用户可以更加全面地捕获和处理数据库中的各种变更事件。

阿里云基于 Flink CDC 提供了一套完整的数据同步解决方案。感兴趣的用户可以通过阿里云平台体验 Flink CDC 的整体数据同步能力。

问答环节

Q1：Flink CDC 3.3 是否支持除 Yarn Session 以外的其他部署模式？

A1：当前版本（3.3）并未计划支持除 Yarn Session 之外的其他部署模式。如果有需要，可以在社区提交 issue ，并在钉钉群中@维护者以获取更多关注和支持。

Q2：OceanBase 的 PIPELINE 目前只支持 MySQL 模式吗？

A2：具体的支持情况建议查阅 Flink 3.3 版本中更新的 OceanBase 连接器使用文档，以确认具体支持的配置。

Q3：MySQL 到 Doris 的整库同步中，MySQL 表无主键且更新关键字不一致时，YAML 配置文件如何配置？

A3：可以通过 Transform 模块指定无主键表的主键。在 Transform 模块中支持使用 primary_key 转换表的 Schema，用户可在配置中指定无主键表转换后的主键。

Q4：Pipeline 是否支持 Oracle Source？

A4：目前 Pipeline 尚未支持 Oracle Source 。由于 Oracle CDC Source 的稳定性有待增强，建议通过 OGG（Oracle GoldenGate）方式使用。

Q5：Pipeline 是否支持 MySQL 到 MySQL 的整库同步？

A5：预计在 3.4 版本中通过 JDBC Pipeline Sink 支持 MySQL 到 MySQL 的同步。目前社区已有计划支持 JDBC Sink 功能。

Q6：全量阶段原表发生 DDL 变更（如加字段）是否支持？

A6：目前不支持。全量阶段使用固定的 Schema 读取数据，表结构变更会导致报错。

Q7：3.3 版本是否支持表级采集流速监控？

A7：

支持情况：在全量阶段支持表级流量监控，通过内置指标实现。监控指标：numRecordsIn：监控同步的数据条数。numBytesIn：监控同步的数据字节量。

Q8：Flink CDC 最新版本是否能动态加表？

A8：Flink 3.3 版本支持动态加表，分为两种场景：

读取全量数据：通过配置 scan.newly-added-tables.enable ，修改需要捕获的表后，从保存点重启即可完成新增表的同步。无需同步历史数据：通过配置 scan.binlog.newly-added-tables.enable，自动解析 create table 语句，同步新增表的变更数据。

A9：目前 pipeline 中仅支持 MySQL。

Q10：如何排查 Flink CDC 作业的问题？

A10：可以在用户群中反馈具体问题，例如在哪些算子或转换环节需要添加日志。框架支持添加 debug 日志，用户可以通过开启日志的方式进行问题排查。

Q11：是否有计划支持国产数据库？

A11：目前社区主要支持较多的数据库如 MySQL 。对于国产数据库，如果没有特定修改且兼容 MySQL 协议，则理论上可以使用现有的连接器。如果有特定需求或修改，暂时没有明确的支持计划。

Q12：Flink CDC 3.3 对 DB2 的支持如何？

A12：虽然 Pipeline 连接器中未包含 DB2，但 CDC Source 是支持 DB2 的。

Q13：如果需要重新抽取某一张表，应该怎么做？

A13：建议开发一个新的作业来进行同步，因为已有作业已存在状态信息，不会重新读取该表的数据。

以上就是本次分享的内容，谢谢大家。