本文作者为数新智能高级架构专家李斌松，曾在科大讯飞、阿里、同盾等公司负责过大数据平台的架构设计与实现。某省医保局项目对多地实时数据采集的要求较高，数新智能调研后确定Flink CDC YAML作业能够满足大部分的业务场景，研发团队结合Flink CDC YAML作业的相关特性，在社区版本基础上进行改造与优化，最终取得节省计算资源高达90%以上、数据实时同步达到秒级延迟的效果。
01
—
项目背景
某省医疗保障局使用省级大数据平台统筹该省所有地市的医疗保障。在原有的平台架构下，省医疗保障局将所有地市以及省级的数据汇聚后以T+1的方式提供给国家平台。如下图所示：
历史数据同步链路
但随着对实时数据的要求越来越高，省医疗保障局之前采购的离线平台以及实时平台，遇到了多条数据采集链路从各地市采集数据，造成各地市数据库压力大，不同采集链路存在数据一致性的问题。每个地市需要同步数据到备份库，导致地市的数据维护工作量巨大。基于以上痛点，某省医疗保障局决定对现有平台进行升级改造。
02
—
架构设计
Flink在国内有非常广泛的应用基础，我们在过往实施的项目中，在实时计算框架选型上会优先考虑Flink相关的技术栈，并已经在使用Flink CDC作为实时数据入湖工具上积累了比较成熟的实施经验。Flink CDC 3.0版本引入了Pipeline的架构设计，功能更为完善，聚焦于数据集成领域，通过YAML作业支持了全增量一体化、整库同步、Schema Evolution、Transform等能力。结合省医疗保障局计划使用Paimon作为大数据存储的背景，Flink YAML作业是满足我们实时数据入湖场景要求的理想方案。
某省医疗保障局的项目中，各地市需要同步的数据来自于地市的多个系统，涉及700+数据库实例，500+逻辑表，其中大部分表均是分库分表形式，分布在多个数据库实例中。同一份数据，需要写入省医保平台的多个存储系统，包括：数据湖平台Paimon、GaussDB(DWS)数据库、TBase数据库、TDSQL数据库等。如下图所示：
新的数据平台架构
如果端到端的数据采集完全依赖MySQL作为数据源，需要多次采集地市MySQL的Binlog，这会带来数据库服务的负担。因此我们引入了Kafka作为中间存储，各个地级市的MySQL库数据先采集到Kafka，再从Kafka读取数据写入到Paimon数据湖存储和MPP数据库。
03
—
技术挑战
由于目前Flink CDC YAML作业对于客户环境的一些数据源类型、实时同步场景尚未支持，无法直接满足该项目的实时同步业务场景。我们在社区Flink CDC 3.1的版本上也做了定制化的二次开发。包括以下内容：
3.1 Kafka Sink输出表结构变更
技术挑战
Flink CDC将原始MySQL DDL SQL解析生成表结构变更事件（SchemaChangeEvent），在Sink侧已经没有原始的DDL SQL语句，并且Kafka Sink没有把这个表结构变更事件发送出来，因此下游没有办法直接通过Kafka的输出获取到上游表结构的变化信息。
解决方案
为了解决这个问题，我们把表结构变更事件直接转为JSON字符串并通过Kafka Pipeline写出，在下游系统消费Kafka Topic读到表结构变更事件对应的JSON后，再反序列为表结构变更事件，下游系统据此执行表结构变更操作。
因为SchemaChangeEvent的子类没有默认构造函数，为了能够序列化或者反序列化，我们在SchemaChangeEvent 及其子类上添加可以用于JSON序列化的注解。代码如下。
SchemaChangeEvent上的改动
3.2 减少作业数
由于医保业务会产生海量的C端数据，而且每个地市的MySQL业务表基本都是分库分表分实例部署模式，总共有470+数据源。为了减少Flink CDC YAML作业的数量和资源消耗，需要多个数据源合并在同一个任务里采集Binlog，并按照实例分发。
解决方案
我们扩展了Flink CDC的Composer模块以支持配置多个DataSource，每个DataSource创建一个DataStream，最后合并所有的DataSource，使得只需要一个作业就能够完成多个实例的数据同步。改动代码如下：
Composer模块的改动
3.3 分库分表合并
技术挑战
不同地市同一业务表数据采集写入同一Topic 中，通过Transform 添加地市编码，再消费Kafka数据入湖。由于地市同一业务表结构有一些微小差别，例如字段类型、字段个数、字段顺序不一致。而Flink CDC里的数据变更事件（DataChangeEvent）不包含表结构信息，需要通过中心节点（SchemaRegistry）进行维护，实现复杂度比较高，在Flink CDC 3.1版本里在包含Transform的分库分表合并场景下的支持并不完善，我们希望能够以更简单的方式支持表结构变更的场景。
解决方案
为解决这个问题，我们在数据变更事件中添加Schema信息，在Transform和Sink 处理时就能直接确定表结构信息, 从而减少与其他组件的交互，降低表结构信息维护的复杂度。
这个改动在Transform和Sink上需要调整的代码比较多，也是我们在实现方案的过程中，开发难度最大的地方。下图是Paimon Sink 改动的方案：
Paimon Sink的改动
3.4 其他新增功能
支持逻辑删除
我们在Sink节点里添加了定义逻辑删除字段的配置，对于上游的删除操作，在配置了这个参数以后，不会实际进行数据删除，而是设置这个逻辑删除字段为 true 作为标识，从而保证所有数据变更都能够被完整记录，方便对医疗数据进行归档和数据回放。下图展示了在YAML作业如何配置一个逻辑删除字段：
配置逻辑删除字段
脏数据管理
在Transform转换阶段以及Sink写入阶段都有可能出现失败的情况，对于这些同步过程中因为类型或者数值限制导致处理失败的数据，我们定义为脏数据，脏数据的出现会导致任务失败，从而使得延迟持续上升。
在出现脏数据时，我们会将这些脏数据写入预先定义好的Kafka Topic，避免因为数据异常导致任务写入失败，同时我们在Flink CDC的事件里记录数据来源的数据源和作业信息，方便后续分析。为此在采集MySQL数据时，我们在数据变更事件里添加数据源地址信息和任务信息，如下图所示，展示了在 Kafka里存储的一条脏数据的内容。
脏数据信息展示
支持MySQL/PostgreSQL Sink
在某省医保局也存在一些使用 MySQL/PostgreSQL 作为存储的业务，为了支持将数据同步到这些存储，我们也拓展支持了实现YAML作业的MySQL/PostgreSQL Sink。
总结
目前这些改动正在逐步贡献给社区，我们希望能够与社区一起推动Flink CDC YAML作业的快速发展。例如在CDC 3.1版本刚刚支持Paimon Sink, 我们很快就开始验证使用了，并且积极和社区同学配合推动完善Paimon Sink实现，经过几次迭代，Paimon Sink在我们线上环境的应用已经比较稳定了。
04
—
实施效果
截止目前，浙江数新网络有限公司（简称：数新智能）CyberData平台已在某省医保局完成Flink CDC YAML作业的生产实践落地，平稳支持来自于10多个地市的700+数据库实例，500+逻辑表，15000+物理表的实时数据同步场景。为支持上述同步场景，总共配置500+ Flink CDC YAML作业，每个任务1 Core 2 GB * 2个并发，实际需要500 Core 1000 GB，通过Flink CDC YAML作业实现整库同步、分库分表同步。此项目实际得到以下多方面的效益：
大幅提升资源利用率：通过YAML作业领先的数据实时同步架构，大幅降低用于数据同步的计算资源消耗。预计可比Flink SQL作业节省Flink集群计算资源高达90%以上。
减少维护成本：通过统一一套数据平台，各地市取消备份库，降低各地市备份库维护人力和物力投入。
提升数据治理水平：通过平台引入的数据比对技术，数据的一致性得到有效保障。通过脏数据收集，能够对异常数据进行追踪。
提升数据实时同步能力：从地市到省级再到国家中心，均实现查询实时数据能力，响应延迟达到秒级别。
欢迎大家多多关注 Flink CDC，从钉钉用户交流群[1]、微信公众号[2]、Slack 频道[3]、邮件列表[4]加入 CDC 用户社区，以及在 Flink CDC GitHub 仓库[5]上参与代码贡献！
[1] “ Flink CDC 社区 ② 群”群的钉钉群号：80655011780
[2] ” Flink CDC 公众号“的微信号：
[3] https://flink.apache.org/what-is-flink/community/#slack
[4] https://flink.apache.org/what-is-flink/community/#mailing-lists
[5] https://github.com/apache/flink-cdc
