这篇文章是开始的时候写了篇随笔，更深刻的cdc的使用和源码分析请参考：深刻解读flink sql cdc的使用以及源码分析html
- CDC简介
- Canal
- CanalJson反序列化源码解析
CDC简介
CDC,Change Data Capture,变动数据获取的简称，使用CDC咱们能够从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变动能够包括INSERT,DELETE,UPDATE等，mysql
用户能够在如下的场景下使用CDC：sql
- 使用flink sql进行数据同步,能够将数据从一个数据同步到其余的地方，好比mysql、elasticsearch等。
- 能够在源数据库上实时的物化一个聚合视图
- 由于只是增量同步，因此能够实时的低延迟的同步数据
- 使用EventTime join 一个temporal表以即可以获取准确的结果
flink 1.11 将这些changelog提取并转化为table apa和sql，目前支持两种格式：Debezium和Canal，这就意味着源表不单单是append操做，并且还有upsert、delete操做。数据库
Canal
接下来咱们使用canal为例简单介绍下CDC的使用apache
canal 格式：json
{
"data": [
{
"id": "13",
"username": "13",
"password": "6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9",
"name": "Canal Manager V2"
}
],
"old": [
{
"id": "13",
"username": "13",
"password": "6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9",
"name": "Canal Manager"
}
],
"database": "canal_manager",
"es": 1568972368000,
"id": 11,
"isDdl": false,
"mysqlType": {...},
"pkNames": [
"id"
],
"sql": "",
"sqlType": {...},
"table": "canal_user",
"ts": 1568972369005,
"type": "UPDATE"
}
简单讲下几个核心的字段:bootstrap
- type : 描述操做的类型，包括‘UPDATE’, 'INSERT', 'DELETE'。
- data : 表明操做的数据。若是为'INSERT'，则表示行的内容；若是为'UPDATE'，则表示行的更新后的状态；若是为'DELETE'，则表示删除前的状态。
- old ：可选字段，若是存在，则表示更新以前的内容，若是不是update操做，则为 null。
完整的语义以下;api
private String destination; // 对应canal的实例或者MQ的topic
private String groupId; // 对应mq的group id
private String database; // 数据库或schema
private String table; // 表名
private List<String> pkNames;
private Boolean isDdl;
private String type; // 类型: INSERT UPDATE DELETE
// binlog executeTime
private Long es; // 执行耗时
// dml build timeStamp
private Long ts; // 同步时间
private String sql; // 执行的sql, dml sql为空
private List<Map<String, Object>> data; // 数据列表
private List<Map<String, Object>> old; // 旧数据列表, 用于update, size和data的size一一对应
-- 定义的字段和data 里面的数据想匹配 CREATE TABLE my_table ( id BIGINT, name STRING, description STRING, weight DECIMAL(10, 2) ) WITH ( 'connector' = 'kafka', 'topic' = 'products_binlog', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'testGroup', 'canal-json.ignore-parse-errors'='true' -- 忽略解析错误，缺省值false );
CanalJson反序列化源码解析
canal 格式也是做为一种flink的格式，并且是source，因此也就是涉及到读取数据的时候进行反序列化，咱们接下来就简单看看CanalJson的反序列化的实现。具体的实现类是CanalJsonDeserializationSchema。数组
咱们看下这个最核心的反序列化方法：app
@Override
public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
try {
//使用json反序列化器将message反序列化成RowData
RowData row = jsonDeserializer.deserialize(message);
//获取type字段，用于下面的判断
String type = row.getString(2).toString();
if (OP_INSERT.equals(type)) {
// 若是操做类型是insert，则data数组表示的是要插入的数据，则循环遍历data，而后添加一个标识INSERT，构造RowData对象，发送下游。
ArrayData data = row.getArray(0);
for (int i = 0; i < data.size(); i++) {
RowData insert = data.getRow(i, fieldCount);
insert.setRowKind(RowKind.INSERT);
out.collect(insert);
}
} else if (OP_UPDATE.equals(type)) {
// 若是是update操做，从data字段里获取更新后的数据、
ArrayData data = row.getArray(0);
// old字段获取更新以前的数据
ArrayData old = row.getArray(1);
for (int i = 0; i < data.size(); i++) {
// the underlying JSON deserialization schema always produce GenericRowData.
GenericRowData after = (GenericRowData) data.getRow(i, fieldCount);
GenericRowData before = (GenericRowData) old.getRow(i, fieldCount);
for (int f = 0; f < fieldCount; f++) {
if (before.isNullAt(f)) {
//若是old字段非空，则说明进行了数据的更新，若是old字段是null，则说明更新先后数据同样，这个时候把before的数据也设置成after的，也就是发送给下游的before和after数据同样。
before.setField(f, after.getField(f));
}
}
before.setRowKind(RowKind.UPDATE_BEFORE);
after.setRowKind(RowKind.UPDATE_AFTER);
//把更新先后的数据都发送下游
out.collect(before);
out.collect(after);
}
} else if (OP_DELETE.equals(type)) {
// 若是是删除操做，data字段里包含将要被删除的数据，把这些数据组织起来发送给下游
ArrayData data = row.getArray(0);
for (int i = 0; i < data.size(); i++) {
RowData insert = data.getRow(i, fieldCount);
insert.setRowKind(RowKind.DELETE);
out.collect(insert);
}
} else {
if (!ignoreParseErrors) {
throw new IOException(format(
"Unknown \"type\" value \"%s\". The Canal JSON message is '%s'", type, new String(message)));
}
}
} catch (Throwable t) {
// a big try catch to protect the processing.
if (!ignoreParseErrors) {
throw new IOException(format(
"Corrupt Canal JSON message '%s'.", new String(message)), t);
}
}
}
参考资料：
[1].https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=147427289
[2].https://flink.apache.org/news/2020/07/06/release-1.11.0.html#table-apisql-support-for-change-data-capture-cdc
更多内容，欢迎关注个人公众号【大数据技术与应用实战】