1. Flink CDC 演进历程Flink CDC 是基于数据库日志 CDCChange Data Capture技术的实时数据集成框架配合 Flink 优秀的管道能力和丰富的上下游生态Flink CDC 可以高效实现海量数据的实时集成。在 2020 年 7 月Flink CDC 作为一个基于个人兴趣孵化的项目合并了第一个 commit拉开了 Flink CDC 实时数据集成的篇章让用户只创建一个简单的 Flink SQL 作业就能完成 CDC 数据的同步、加工和分析。这个阶段里存在通过加锁保证一致性并且不支持水平拓展的问题Flink CDC 参考 DBLog 论文 的设计实现了无锁并发读取的全增量同步完成了从 1.0 到 2.0 的升级。Flink CDC 2.0 受到了广大用户的好评不过在广泛应用的过程中也暴露出了一些有待提升的地方需要提升的部分主要包括通过 SQL 定义表结构的方式在上游表发生加减列时需要手动调整作业在整库同步的场景下需要为每一张表创建一个作业占用连接多计算资源消耗大等。在社区用户与开发者的共同努力下Flink CDC 于 2023 年 12 月完成了 3.0 版本[1] 的功能落地提供了强大的端到端的全增量同步、表结构变更自动同步、整库同步、分库分表同步等高级特性有效地解决了用户的痛点。2. Flink CDC 3.0 的架构设计Flink CDC 3.0 的核心特性包括端到端数据集成用户只需要配置一个 YAML 文件就能快速构建数据入湖入仓作业完整的数据同步全量读取结束自动同步增量数据并且上游表结构变更自动应用到下游一个作业实例支持读取和写入多表占用数据库连接少增量读取阶段自动关闭空闲读取器节省计算资源Flink CDC 3.0 的整体架构自顶而下分为 4 层Flink CDC API面向终端用户的 API 层用户使用 YAML 格式配置数据同步流水线使用 Flink CDC CLI 提交任务Flink CDC Connect对接外部系统的连接器层通过对 Flink 与现有 Flink CDC source 进行封装实现对外部系统同步数据的读取和写入Flink CDC Composer同步任务的构建层将用户的同步任务翻译为 Flink DataStream 作业Flink CDC Runtime运行时层根据数据同步场景高度定制 Flink 算子实现 schema 变更、路由、变换等高级功能3. Flink CDC 3.0 的核心实现3.1 数据抽象Event 是 Flink CDC 3.0 内部进行数据处理及传输的数据结构接口其作用类似于 Flink SQL 中的 RowData 接口。Event 目前所有的实现如下图所示。3.1.1 ChangeEventChangeEvent 接口代表着在一张表上发生过的变更事件实现类包括数据变更事件即 DataChangeEvent 类)和表结构变更事件即继承 SchemaChangeEvent 接口的类两种DataChangeEvent 里保存了完整的数据变更信息即包含了变更前before和变更后after每条记录的字段值SchemaChangeEvent 有增加列、删除列、修改列类型等实现。Flink CDC 把表结构变更信息当成一种事件流转这样能够避免在数据变更事件里保存类型信息需要从 DataChangeEvent 读取数据的节点会基于 SchemaChangeEvent 维护表结构信息。Flink CDC 还实现了自己的序列化器每条记录使用二进制的方式存储在 Flink 的 MemorySegment 中通过这种底层结构的优化设计有效提高在不同节点之间数据流转的效率。3.1.2 FlushEventFlushEvent 是包含数据刷写控制逻辑的特殊事件。当发生表结构变更事件后之前的数据可能尚未处理完链路上会并存两种不同表结构的数据。大部分数据库不允许直接在同一批次中混合处理两种表格式的数据在处理新版本的数据之前必须确保旧版本的数据已全部完成刷写操作。FlushEvent 作用是间隔这两种数据在 Sink 端接受到 FlushEvent 后就需要将之前缓存的数据全部刷写出去。3.2 算子编排FlinkCDC 根据数据集成的场景深度定制了 Flink DataStream 的算子链路目前制定的数据处理链路如下图所示下面对这些模块的具体实现做进一步的介绍。3.2.1 SourceSource 模块负责生产在链路中流转的变更事件。FlinkCDC 2.0 提供了强大的全增量同步、并发读取的能力已经能够生成包含各类变更事件信息的 SourceRecord 对象在此基础上只需要再实现一个将 SourceRecord 解析成前面介绍的各种表变更事件的 DebeziumDeserializationSchema 自定义转换器就能完成 FlinkCDC 3.0 数据源的接入。在第一次启动时Source 模块需要先拉取表结构信息并生成 CreateTableEvent 发送到下游中这是为了让下游节点能够解析 DataChangeEvent。publicvoidenterAlterByDropColumn(MySqlParser.AlterByDropColumnContextctx){StringremovedColNameparser.parseName(ctx.uid());changes.add(newDropColumnEvent(currentTable,Collections.singletonList(removedColName)));super.enterAlterByDropColumn(ctx);}3.2.2 Transform在当前版本暂未实现。3.2.3 Schema在发生表结构变更事件以后Schema 模块负责阻塞上游数据的继续发放直到旧版本格式数据刷写完毕。这个逻辑需要通过 FlushEvent 来传递由于下游可能存在多个 Sink需要通过运行在 JobManager 上的一个 OperatorCoordinator 来进行管控这个 OperatorCoordinator 称为 SchemaRegistry。具体来说处理表结构变更的流程如下图所示在程序启动时所有的 Sink Operator 都向 SchemaRegistry 注册SchemaRegistry 记录 Writer 的数量。在收到来自 Source 的 SchemaChangeEvent 时SchemaOperator 发送一个包含本次表结构变更事件的 SchemaChangeRequest 给 SchemaRegistry让 SchemaRegistry 缓存这个 SchemaChangeEvent。SchemaOperator 下发一个 FlushEvent 给所有的 Sink OperatorSink Operator 接收到 FlushEvent 后刷写数据到外部系统并且发送 FlushSuccessEvent 向 SchemaRegistry 进行汇报。SchemaRegistry 据此统计响应的 Writer 数量。SchemaOperator 下发 SchemaChangeEvent 给所有的 Sink Operator让 Sink Operator 更新对应表的序列化器。SchemaOperator 发送一个 ReleaseUpstreamRequest 给 SchemaRegistry并且开始阻塞自身不再处理任何变更事件直到收到 SchemaRegistry 的回应。SchemaRegistry 接收到 FlushSuccessEvent 以后会和第 1 步中注册的 Sink Operator 进行比较如果所有的 Sink Operator 都已刷写完毕则开始将第 2 步中受到的 SchemaChangeEvent 应用到外部系统中并且对第 4 步接收到的 ReleaseUpstreamRequest 进行回应。这样SchemaOperator 就可以开始继续传递新的数据变更时和表结构变更事件了。3.2.4 RouteRoute 模块提供了表名映射的能力。通过为每一个源表中的数据设置其写入的目标表通过一对一以及多对一的映射配置我们能够实现整库同步和简单的分库分表同步功能。Route 模块基于 Flink 的 RichMapFunction 实现允许通过 source-table 指定一个正则表达式规则将一系列符合正则表达式规则的表名替换到另外一个由 sink-table 指定的表名。RouteFunction 的核心代码如下publicEventmap(Eventevent)throwsException{ChangeEventchangeEvent(ChangeEvent)event;TableIdtableIdchangeEvent.tableId();for(Tuple2Selectors,TableIdroute:routes){Selectorsselectorsroute.f0;TableIdreplaceByroute.f1;if(selectors.isMatch(tableId)){returnrecreateChangeEvent(changeEvent,replaceBy);}}returnevent;}3.2.5 Partition在数据同步场景数据的生产和消费的速率常常是不匹配的用户希望能够通过增加 Sink 的并发度来提高数据处理的速率。Partition 模块负责分发事件到不同的 Sink 中。在 Partition 阶段数据变更事件按照表名和主键作为哈希键保证同一张表中相同主键的数据不会因数据分发出现乱序的情况。哈希键的计算方式如下所示publicIntegerapply(DataChangeEventevent){ListObjectobjectsToHashnewArrayList();// Table IDTableIdtableIdevent.tableId();Optional.ofNullable(tableId.getNamespace()).ifPresent(objectsToHash::add);Optional.ofNullable(tableId.getSchemaName()).ifPresent(objectsToHash::add);objectsToHash.add(tableId.getTableName());// Primary keyRecordDatadataevent.op().equals(OperationType.DELETE)?event.before():event.after();for(RecordData.FieldGetterprimaryKeyGetter:primaryKeyGetters){objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));}// Calculate hashreturn(Objects.hash(objectsToHash.toArray())*31)0x7FFFFFFF;}同时由于 Sink 模块需要维护表结构信息对于表结构变更事件需要广播到每一个并发里。对于控制数据刷写的 FlushEvent也需要广播到每一个下游的每一个通道里。其代码实现如下publicvoidprocessElement(StreamRecordEventelement)throwsException{Eventeventelement.getValue();if(eventinstanceofSchemaChangeEvent){// Update hash functionTableIdtableId((SchemaChangeEvent)event).tableId();cachedHashFunctions.put(tableId,recreateHashFunction(tableId));// Broadcast SchemaChangeEventbroadcastEvent(event);}elseif(eventinstanceofFlushEvent){// Broadcast FlushEventbroadcastEvent(event);}elseif(eventinstanceofDataChangeEvent){// Partition DataChangeEvent by table ID and primary keyspartitionBy(((DataChangeEvent)event));}}3.2.6 Sink在 Sink 模块需要将数据写出到外部系统中并且将表结构变更应用到外部系统中。Flink CDC 的 DataSink API 提供了 EventSinkProvider 和 MetaDataApplier 接口去完成这两件事情。EventSinkProvider 用于将表数据变更应用到外部系统中。EventSinkProvider 要求提供一个基于 Flink SinkFunction 或者是 Flink Sink API 的实现并且具备写出到多个表的能力。以 Flink Sink API 为例SinkWriter 需要从 DataChangeEvent 中取出变更数据并写出到对应的表中。当处理到 SchemaChangeEvent 时 SinkWriter 更新内存中保存的表结构信息。当处理到 FlushEvent 时 Sink Operator 会调用 SinkWriter 的 flush 方法将数据刷写出去。MetaDataApplier 用于将表结构变更应用到外部系统中。在 SchemaRegistry 接受到所有的 Sink 算子处理完 FlushEvent 的通知后由 SchemaRegistry 负责调用 MetaDataApplier 的 applySchemaChange方法去应用表结构变更事件。考虑到任务重启的情况MetaDataApplier 需要支持对一个表结构变更事件幂等处理。4. 未来规划Flink CDC 社区致力于持续深化数据同步与处理的全面性和灵活性。在 Flink CDC 3.0 里针对数据集成场景定制了高效的数据格式和算子编排实现了对表结构变更同步和整库同步的支持。基于未来的演进规划社会将会着重关注完善 Transform 模块的功能以满足用户对数据同步过程中的深度定制需求。计划在下一个大版本中支持表结构动态调整包括裁剪不必要的列、添加计算列等功能以及提供数据过滤能力让用户能够在同步过程中一站式完成复杂的数据转换任务。此外在连接器生态建设方面社区正着手接入 Kafka、PostgreSQL 等业界主流的数据源以及 Paimon、Iceberg 等先进的湖仓存储系统。进一步拓宽 Flink CDC 的上下游数据集成范围推动上下游组件的深度融合。
1. Flink CDC 演进历程Flink CDC 是基于数据库日志 CDCChange Data Capture技术的实时数据集成框架配合 Flink 优秀的管道能力和丰富的上下游生态Flink CDC 可以高效实现海量数据的实时集成。在 2020 年 7 月Flink CDC 作为一个基于个人兴趣孵化的项目合并了第一个 commit拉开了 Flink CDC 实时数据集成的篇章让用户只创建一个简单的 Flink SQL 作业就能完成 CDC 数据的同步、加工和分析。这个阶段里存在通过加锁保证一致性并且不支持水平拓展的问题Flink CDC 参考 DBLog 论文 的设计实现了无锁并发读取的全增量同步完成了从 1.0 到 2.0 的升级。Flink CDC 2.0 受到了广大用户的好评不过在广泛应用的过程中也暴露出了一些有待提升的地方需要提升的部分主要包括通过 SQL 定义表结构的方式在上游表发生加减列时需要手动调整作业在整库同步的场景下需要为每一张表创建一个作业占用连接多计算资源消耗大等。在社区用户与开发者的共同努力下Flink CDC 于 2023 年 12 月完成了 3.0 版本[1] 的功能落地提供了强大的端到端的全增量同步、表结构变更自动同步、整库同步、分库分表同步等高级特性有效地解决了用户的痛点。2. Flink CDC 3.0 的架构设计Flink CDC 3.0 的核心特性包括端到端数据集成用户只需要配置一个 YAML 文件就能快速构建数据入湖入仓作业完整的数据同步全量读取结束自动同步增量数据并且上游表结构变更自动应用到下游一个作业实例支持读取和写入多表占用数据库连接少增量读取阶段自动关闭空闲读取器节省计算资源Flink CDC 3.0 的整体架构自顶而下分为 4 层Flink CDC API面向终端用户的 API 层用户使用 YAML 格式配置数据同步流水线使用 Flink CDC CLI 提交任务Flink CDC Connect对接外部系统的连接器层通过对 Flink 与现有 Flink CDC source 进行封装实现对外部系统同步数据的读取和写入Flink CDC Composer同步任务的构建层将用户的同步任务翻译为 Flink DataStream 作业Flink CDC Runtime运行时层根据数据同步场景高度定制 Flink 算子实现 schema 变更、路由、变换等高级功能3. Flink CDC 3.0 的核心实现3.1 数据抽象Event 是 Flink CDC 3.0 内部进行数据处理及传输的数据结构接口其作用类似于 Flink SQL 中的 RowData 接口。Event 目前所有的实现如下图所示。3.1.1 ChangeEventChangeEvent 接口代表着在一张表上发生过的变更事件实现类包括数据变更事件即 DataChangeEvent 类)和表结构变更事件即继承 SchemaChangeEvent 接口的类两种DataChangeEvent 里保存了完整的数据变更信息即包含了变更前before和变更后after每条记录的字段值SchemaChangeEvent 有增加列、删除列、修改列类型等实现。Flink CDC 把表结构变更信息当成一种事件流转这样能够避免在数据变更事件里保存类型信息需要从 DataChangeEvent 读取数据的节点会基于 SchemaChangeEvent 维护表结构信息。Flink CDC 还实现了自己的序列化器每条记录使用二进制的方式存储在 Flink 的 MemorySegment 中通过这种底层结构的优化设计有效提高在不同节点之间数据流转的效率。3.1.2 FlushEventFlushEvent 是包含数据刷写控制逻辑的特殊事件。当发生表结构变更事件后之前的数据可能尚未处理完链路上会并存两种不同表结构的数据。大部分数据库不允许直接在同一批次中混合处理两种表格式的数据在处理新版本的数据之前必须确保旧版本的数据已全部完成刷写操作。FlushEvent 作用是间隔这两种数据在 Sink 端接受到 FlushEvent 后就需要将之前缓存的数据全部刷写出去。3.2 算子编排FlinkCDC 根据数据集成的场景深度定制了 Flink DataStream 的算子链路目前制定的数据处理链路如下图所示下面对这些模块的具体实现做进一步的介绍。3.2.1 SourceSource 模块负责生产在链路中流转的变更事件。FlinkCDC 2.0 提供了强大的全增量同步、并发读取的能力已经能够生成包含各类变更事件信息的 SourceRecord 对象在此基础上只需要再实现一个将 SourceRecord 解析成前面介绍的各种表变更事件的 DebeziumDeserializationSchema 自定义转换器就能完成 FlinkCDC 3.0 数据源的接入。在第一次启动时Source 模块需要先拉取表结构信息并生成 CreateTableEvent 发送到下游中这是为了让下游节点能够解析 DataChangeEvent。publicvoidenterAlterByDropColumn(MySqlParser.AlterByDropColumnContextctx){StringremovedColNameparser.parseName(ctx.uid());changes.add(newDropColumnEvent(currentTable,Collections.singletonList(removedColName)));super.enterAlterByDropColumn(ctx);}3.2.2 Transform在当前版本暂未实现。3.2.3 Schema在发生表结构变更事件以后Schema 模块负责阻塞上游数据的继续发放直到旧版本格式数据刷写完毕。这个逻辑需要通过 FlushEvent 来传递由于下游可能存在多个 Sink需要通过运行在 JobManager 上的一个 OperatorCoordinator 来进行管控这个 OperatorCoordinator 称为 SchemaRegistry。具体来说处理表结构变更的流程如下图所示在程序启动时所有的 Sink Operator 都向 SchemaRegistry 注册SchemaRegistry 记录 Writer 的数量。在收到来自 Source 的 SchemaChangeEvent 时SchemaOperator 发送一个包含本次表结构变更事件的 SchemaChangeRequest 给 SchemaRegistry让 SchemaRegistry 缓存这个 SchemaChangeEvent。SchemaOperator 下发一个 FlushEvent 给所有的 Sink OperatorSink Operator 接收到 FlushEvent 后刷写数据到外部系统并且发送 FlushSuccessEvent 向 SchemaRegistry 进行汇报。SchemaRegistry 据此统计响应的 Writer 数量。SchemaOperator 下发 SchemaChangeEvent 给所有的 Sink Operator让 Sink Operator 更新对应表的序列化器。SchemaOperator 发送一个 ReleaseUpstreamRequest 给 SchemaRegistry并且开始阻塞自身不再处理任何变更事件直到收到 SchemaRegistry 的回应。SchemaRegistry 接收到 FlushSuccessEvent 以后会和第 1 步中注册的 Sink Operator 进行比较如果所有的 Sink Operator 都已刷写完毕则开始将第 2 步中受到的 SchemaChangeEvent 应用到外部系统中并且对第 4 步接收到的 ReleaseUpstreamRequest 进行回应。这样SchemaOperator 就可以开始继续传递新的数据变更时和表结构变更事件了。3.2.4 RouteRoute 模块提供了表名映射的能力。通过为每一个源表中的数据设置其写入的目标表通过一对一以及多对一的映射配置我们能够实现整库同步和简单的分库分表同步功能。Route 模块基于 Flink 的 RichMapFunction 实现允许通过 source-table 指定一个正则表达式规则将一系列符合正则表达式规则的表名替换到另外一个由 sink-table 指定的表名。RouteFunction 的核心代码如下publicEventmap(Eventevent)throwsException{ChangeEventchangeEvent(ChangeEvent)event;TableIdtableIdchangeEvent.tableId();for(Tuple2Selectors,TableIdroute:routes){Selectorsselectorsroute.f0;TableIdreplaceByroute.f1;if(selectors.isMatch(tableId)){returnrecreateChangeEvent(changeEvent,replaceBy);}}returnevent;}3.2.5 Partition在数据同步场景数据的生产和消费的速率常常是不匹配的用户希望能够通过增加 Sink 的并发度来提高数据处理的速率。Partition 模块负责分发事件到不同的 Sink 中。在 Partition 阶段数据变更事件按照表名和主键作为哈希键保证同一张表中相同主键的数据不会因数据分发出现乱序的情况。哈希键的计算方式如下所示publicIntegerapply(DataChangeEventevent){ListObjectobjectsToHashnewArrayList();// Table IDTableIdtableIdevent.tableId();Optional.ofNullable(tableId.getNamespace()).ifPresent(objectsToHash::add);Optional.ofNullable(tableId.getSchemaName()).ifPresent(objectsToHash::add);objectsToHash.add(tableId.getTableName());// Primary keyRecordDatadataevent.op().equals(OperationType.DELETE)?event.before():event.after();for(RecordData.FieldGetterprimaryKeyGetter:primaryKeyGetters){objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));}// Calculate hashreturn(Objects.hash(objectsToHash.toArray())*31)0x7FFFFFFF;}同时由于 Sink 模块需要维护表结构信息对于表结构变更事件需要广播到每一个并发里。对于控制数据刷写的 FlushEvent也需要广播到每一个下游的每一个通道里。其代码实现如下publicvoidprocessElement(StreamRecordEventelement)throwsException{Eventeventelement.getValue();if(eventinstanceofSchemaChangeEvent){// Update hash functionTableIdtableId((SchemaChangeEvent)event).tableId();cachedHashFunctions.put(tableId,recreateHashFunction(tableId));// Broadcast SchemaChangeEventbroadcastEvent(event);}elseif(eventinstanceofFlushEvent){// Broadcast FlushEventbroadcastEvent(event);}elseif(eventinstanceofDataChangeEvent){// Partition DataChangeEvent by table ID and primary keyspartitionBy(((DataChangeEvent)event));}}3.2.6 Sink在 Sink 模块需要将数据写出到外部系统中并且将表结构变更应用到外部系统中。Flink CDC 的 DataSink API 提供了 EventSinkProvider 和 MetaDataApplier 接口去完成这两件事情。EventSinkProvider 用于将表数据变更应用到外部系统中。EventSinkProvider 要求提供一个基于 Flink SinkFunction 或者是 Flink Sink API 的实现并且具备写出到多个表的能力。以 Flink Sink API 为例SinkWriter 需要从 DataChangeEvent 中取出变更数据并写出到对应的表中。当处理到 SchemaChangeEvent 时 SinkWriter 更新内存中保存的表结构信息。当处理到 FlushEvent 时 Sink Operator 会调用 SinkWriter 的 flush 方法将数据刷写出去。MetaDataApplier 用于将表结构变更应用到外部系统中。在 SchemaRegistry 接受到所有的 Sink 算子处理完 FlushEvent 的通知后由 SchemaRegistry 负责调用 MetaDataApplier 的 applySchemaChange方法去应用表结构变更事件。考虑到任务重启的情况MetaDataApplier 需要支持对一个表结构变更事件幂等处理。4. 未来规划Flink CDC 社区致力于持续深化数据同步与处理的全面性和灵活性。在 Flink CDC 3.0 里针对数据集成场景定制了高效的数据格式和算子编排实现了对表结构变更同步和整库同步的支持。基于未来的演进规划社会将会着重关注完善 Transform 模块的功能以满足用户对数据同步过程中的深度定制需求。计划在下一个大版本中支持表结构动态调整包括裁剪不必要的列、添加计算列等功能以及提供数据过滤能力让用户能够在同步过程中一站式完成复杂的数据转换任务。此外在连接器生态建设方面社区正着手接入 Kafka、PostgreSQL 等业界主流的数据源以及 Paimon、Iceberg 等先进的湖仓存储系统。进一步拓宽 Flink CDC 的上下游数据集成范围推动上下游组件的深度融合。
双梁正问题 【 torch 实战】（Python代码实现）)
