赞
踩
CDC,Change Data Capture,变更数据获取的简称,使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游,供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等,
用户可以在以下的场景下使用CDC:
flink 1.11 将这些changelog提取并转化为table apa和sql,目前支持两种格式:Debezium和Canal,这就意味着源表不仅仅是append操作,而且还有upsert、delete操作。
接下来我们使用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"
- }
简单讲下几个核心的字段:
完整的语义如下;
- 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
- );
canal 格式也是作为一种flink的格式,而且是source,所以也就是涉及到读取数据的时候进行反序列化,我们接下来就简单看看CanalJson的反序列化的实现。具体的实现类是CanalJsonDeserializationSchema。
我们看下这个最核心的反序列化方法:
- @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
更多内容,欢迎关注我的公众号【大数据技术与应用实战】
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。