flink 1.11中的CDC是什么意思

本篇文章给大家分享的是有关flink 1.11中的CDC是什么意思，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。

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。

完整的语义如下;

    private String                    destination;                            // 对应canal的实例或者MQ的topic    private String                    groupId;                                // 对应mq的group id    private String                    database;                               // 数据库或schema    private String                    table;                                  // 表名    private List              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> data;                                   // 数据列表    private List> 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。

我们看下这个最核心的反序列化方法：

 @Override        public void deserialize(byte[] message, Collector 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);                        }                }        }

以上就是flink 1.11中的CDC是什么意思，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注行业资讯频道。