CDC是什么?
CDC(Change Data Capture),变更数据获取的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等.
用户可以在如下的场景使用cdc：
-
实时数据同步：比如我们将mysql库中的数据同步到我们的数仓中。
-
数据库的实时物化视图。
可以从下面表格对比他们功能和差异点。
经过以上对比，我们可以发现基于日志 CDC 有以下这几种优势：
- 能够捕获所有数据的变化，捕获完整的变更记录。在异地容灾，数据备份等场景中得到广泛应用，如果是基于查询的 CDC 有可能导致两次查询的中间一部分数据丢失
- 每次 DML 操作均有记录无需像查询 CDC 这样发起全表扫描进行过滤，拥有更高的效率和性能，具有低延迟，不增加数据库负载的优势
- 无需入侵业务，业务解耦，无需更改业务模型
- 捕获删除事件和捕获旧记录的状态，在查询 CDC 中，周期的查询无法感知中间数据是否删除
在实时性、吞吐量方面占优，如果数据源是 MySQL、PostgreSQL、MongoDB 等常见的数据库实现，建议使用 Debezium 来实现变更数据的捕获（下图来自 Debezium 官方文档）。如果使用的只有 MySQL，则还可以用 Canal。 
Flink 内置了 Debezium
Flink-1.11 版本正式发布CDC
Canal 不支持读取全量 binlog 数据，而 Flink CDC 完美避开了这个问题
Flink 社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件
Flink Table / SQL 模块将数据库表和变动记录流（例如 CDC 的数据流）看做是 动态表 (Dynamic Table)，因此内部提供的 Upsert 消息结构（+I 表示新增、-U 表示记录更新前的值、+U 表示记录更新后的值，-D 表示删除）可以与 Debezium 等生成的变动记录一一对应。
目前 Flink CDC 支持两种数据源输入方式。
2.1.1、输入 Debezium 等数据流进行同步例如 MySQL -> Debezium -> Kafka -> Flink -> PostgreSQL。适用于已经部署好了 Debezium，希望暂存一部分数据到 Kafka 中以供多次消费，只需要 Flink 解析并分发到下游的场景。
在该场景下，由于 CDC 变更记录会暂存到 Kafka 一段时间，因此可以在这期间任意启动/重启 Flink 作业进行消费；也可以部署多个 Flink 作业对这些数据同时处理并写到不同的数据目的（Sink）库表中，实现了 Source 变动与 Sink 的解耦。
我们还可以跳过 Debezium 和 Kafka 的中转，使用 Flink CDC Connectors 对上游数据源的变动进行直接的订阅处理。从内部实现上讲，Flink CDC Connectors 内置了一套 Debezium 和 Kafka 组件，但这个细节对用户屏蔽，因此用户看到的数据链路如下图所示： 
flink-json 模块中的
org.apache.flink.formats.json.debezium.DebeziumJsonFormatFactory是负责构造解析Debezium JSON格式的工厂类；org.apache.flink.formats.json.canal.CanalJsonFormatFactory负责Canal JSON格式。
这些类已经内置在 Flink 1.11 的发行版中，直接可以使用，无需附加任何程序包。
对于 Debezium JSON 格式而言，Flink 将具体的解析逻辑放在了org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema#DebeziumJsonDeserializationSchema 类中。
下图表示 Debezium JSON 的一条更新（Update）消息，它表示上游已将 id=123 的数据更新，且字段内包含了更新前的旧值，以及更新后的新值。 
那么，Flink 是如何解析并生成对应的 Flink 消息呢？我们看下这个类的 deserialize 方法：
GenericRowData before = (GenericRowData) payload.getField(0); // 更新前的数据
GenericRowData after = (GenericRowData) payload.getField(1); // 更新后的数据
String op = payload.getField(2).toString(); // 获取 "op" 字段的类型
if (OP_CREATE.equals(op) || OP_READ.equals(op)) { // 如果是创建 (c) 或快照读取 (r) 消息
after.setRowKind(RowKind.INSERT); // 设置消息类型为新建 (+I)
out.collect(after); // 发送给下游
} else if (OP_UPDATE.equals(op)) { // 如果是更新 (u) 消息
before.setRowKind(RowKind.UPDATE_BEFORE); // 把更新前的数据类型设置为撤回 (-U)
after.setRowKind(RowKind.UPDATE_AFTER); // 把更新后的数据类型设置为更新 (+U)
out.collect(before); // 发送两条数据给下游
out.collect(after);
} else if (OP_DELETE.equals(op)) { // 如果是删除 (d) 消息
before.setRowKind(RowKind.DELETE); // 将消息类型设置为删除 (-D)
out.collect(before); // 发送给下游
} else {
... // 异常处理逻辑
}
从上述逻辑可以看出，对于每一种 Debezium 的操作码（op 字段的类型），都可以用 Flink 的 RowKind 类型来表示。对于插入 +I 和删除 D，都只需要一条消息即可；而对于更新，则涉及删除旧数据和写入新数据，因此需要 -U 和 +U 两条消息来对应。
特别地，在 MySQL、PostgreSQL 等支持 Upsert（原子操作的 Update or Insert）语义的数据库中，通常前一个 -U 消息可以省略，只把后一个 +U 消息用作实际的更新操作即可，这个优化在 Flink 中也有实现。
因此可以看到，Debezium 到 Flink 消息的转换逻辑是非常简单和自然的，这也多亏了 Flink 先进的设计理念，很早就提出并实现了 Upsert 数据流和动态数据表之间的映射关系。
3.2、Flink CDC Connectors 的实现 3.2.1、flink-connector-debezium 模块我们在使用 Flink CDC Connectors 时，也会好奇它究竟是如何做到的不需要安装和部署外部服务就可以实现 CDC 的。当我们阅读 flink-connector-mysql-cdc 的源码时，可以看到它内部依赖了 flink-connector-debezium 模块，而这个模块将 Debezium Embedded 嵌入到了 Connector 中。
flink-connector-debezium 的数据源实现类为 com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction，它集成了 Flink 中的 RichSourceFunction 并实现了 CheckpointedFunction 以支持快照保存状态。
通常而言，对于 SourceFunction，我们可以从它的 run 方法入手分析。它的核心代码如下：
this.engine = DebeziumEngine.create(Connect.class)
.using(properties) // 初始化 Debezium 所需的参数
.notifying(debeziumConsumer) // 收到批量的变更消息, 则 Debezium 会回调 DebeziumChangeConsumer 来反序列化并向下游输出数据
.using(OffsetCommitPolicy.always())
.using(
(success, message, error) -> {
if (!success && error != null) {
this.reportError(error);
}
})
.build();
...
executor.execute(engine); // 向 Executor 提交 Debezium 线程以启动运行
可以看到，这个 SourceFunction 使用一些预先定义的参数，初始化了一个嵌入式的 DebeziumEngine（Java 的 Runnable），然后提交给线程池（executor）去执行。这个 Debezium 线程会批量接收 binlog 信息并回调传入的 debeziumConsumer 以反序列化消息并交给 Flink 来处理。本类的其他方法主要负责初始化状态和保存快照，这里略过。
这里我们再来看一下 DebeziumChangeConsumer 的实现，它的最核心的方法是 handleBatch 。当 Debezium 收到一批新的事件时，会调用这个方法来通知我们的 Connector 进行处理。这里有个 for 循环轮询的逻辑：
for (ChangeEvent event : changeEvents) { // 轮询各个事件
SourceRecord record = event.value();
if (isHeartbeatEvent(record)) { // 如果时心跳包
// 只更新当前 offset 信息, 然后继续（不进行实际处理）
synchronized (checkpointLock) {
debeziumOffset.setSourcePartition(record.sourcePartition());
debeziumOffset.setSourceOffset(record.sourceOffset());
}
continue;
}
deserialization.deserialize(record, debeziumCollector); // 反序列化这条消息
if (isInDbSnapshotPhase) { // 如果处于数据库快照期, 需要阻止 Flink 检查点（Checkpoint）生成
if (!lockHold) {
MemoryUtils.UNSAFE.monitorEnter(checkpointLock);
lockHold = true;
...
}
if (!isSnapshotRecord(record)) { // 如果已经不在数据库快照期了, 就释放锁, 允许 Flink 正常生成检查点（Checkpoint）
MemoryUtils.UNSAFE.monitorExit(checkpointLock);
isInDbSnapshotPhase = false;
...
}
}
// 更新当前 offset 信息, 并向下游 Flink 算子发送数据
emitRecordsUnderCheckpointLock(
debeziumCollector.records, record.sourcePartition(), record.sourceOffset());
}
可以看到逻辑比较简单，只需要关注 checkpointLock 这个对象：只有持有这个对象的锁时，才允许 Flink 进行检查点的生成。
当作业处于数据库快照期（即作业刚启动时，需全量同步源数据库的一份完整快照，此时收到的数据类型是 Debezium 的 SnapshotRecord），则不允许 Flink 进行 Checkpoint 即检查点的生成，以避免作业崩溃恢复后状态不一致；同样地，如果正在向下游算子发送数据并更新 offset 信息时，也不允许快照的进行。这些操作都是为了保证 Exacly-Once（精确一致）语义。
这里也解释了在作业刚启动时，如果数据库较大（同步时间较久），Flink 刚开始的 Checkpoint 永远失败（超时）的原因：只有当 Flink 完整同步了全量数据后，才可以进行增量数据的处理，以及 Checkpoint 的生成。
3.2.2、flink-connector-mysql-cdc 模块而对于 flink-connector-mysql-cdc 模块而言，它主要涉及到 MySQLTableSource 的声明和实现。
我们知道，Flink 是通过 Java 的 SPI（Service Provider Interface）机制动态加载 Connector 的，因此我们首先看这个模块的 src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory 文件，里面内容指向 com.alibaba.ververica.cdc.connectors.mysql.table.MySQLTableSourceFactory。
打开这个工厂类，我们可以看到它定义了该 Connector 所需的参数，例如 MySQL 数据库的用户名、密码、表名等信息，并负责 MySQLTableSource 实例的具体创建，而 MySQLTableSource 类对这些参数做转换，最终会生成一个上文提到的 DebeziumSourceFunction 对象。
因此我们可以发现，这个模块作用是一个 MySQL 参数的封装和转换层，最终的逻辑实现仍然是由 flink-connector-debezium 完成的。
关注我的公众号【宝哥大数据】
