文章目录
- CDC简介
- Canal
- CanalJson反序列化源码解析
CDC简介
CDC,Change Data Capture,变更数据获取的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等，
用户可以在以下的场景下使用CDC：
- 使用flink sql进行数据同步,可以将数据从一个数据同步到其他的地方，比如mysql、elasticsearch等。
- 可以在源数据库上实时的物化一个聚合视图
- 因为只是增量同步，所以可以实时的低延迟的同步数据
- 使用EventTime join 一个temporal表以便可以获取准确的结果
flink 1.11 将这些changelog提取并转化为table apa和sql，目前支持两种格式：Debezium和Canal，这就意味着源表不仅仅是append操作，而且还有upsert、delete操作。
Canal
接下来我们使用canal为例简单介绍下CDC的使用
canal 格式：
{"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"
}
简单讲下几个核心的字段:
- type : 描述操作的类型，包括‘UPDATE’, ‘INSERT’, ‘DELETE’。
- data : 代表操作的数据。如果为’INSERT’，则表示行的内容；如果为’UPDATE’，则表示行的更新后的状态；如果为’DELETE’，则表示删除前的状态。
- old ：可选字段，如果存在，则表示更新之前的内容，如果不是update操作，则为 null。
-- 定义的字段和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。
我们看下这个最核心的反序列化方法：
@Overridepublic void deserialize(byte[] message, Collector<RowData> out) throws IOException {try {//使用json反序列化器将message反序列化成RowDataRowData 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
更多内容，欢迎关注我的公众号【大数据技术与应用实战】
