flink教程-详解flink 1.11 中的CDC (Change Data Capture)

{
  "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
);

 @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);
   }
  }
 }