Canal Format

功能描述

Canal是一个 CDC（ChangeLog Data Capture，变更日志数据捕获）工具，可以实时地将 MySQL 变更传输到其他系统。Canal 为变更日志提供了统一的数据格式，并支持使用 JSON 或 protobuf序列化消息（Canal 默认使用 protobuf）。

Flink 支持将 Canal 的 JSON 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如

• 将增量数据从数据库同步到其他系统
• 日志审计
• 数据库的实时物化视图
• 关联维度数据库的变更历史，等等。

Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Canal 格式的 JSON 消息，输出到 Kafka 等存储中。 但需要注意的是，目前 Flink 还不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此，Flink 将 UPDATE_BEFORE 和 UPDATE_AFTER 分别编码为 DELETE 和 INSERT 类型的 Canal 消息。

参数说明

支持的Connector

• Kafka

示例

使用kafka发送数据，输出到print中。

1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列-->找到作业的所属队列-->更多-->测试地址连通性-->输入kafka的地址-->测试）。若能连通，则表示跨源已经绑定成功；否则表示未成功。
2. 创建flink opensource sql作业，选择flink1.12版本，并提交运行，其代码如下：

   create table kafkaSource(
     id bigint,
     name string,
     description string,
     weight DECIMAL(10, 2)
     ) with (
       'connector' = 'kafka',
       'topic' = '<yourTopic>',
       'properties.group.id' = '<yourGroupId>',
       'properties.bootstrap.servers' = '<yourKafkaAddress>:<yourKafkaPort>',
       'scan.startup.mode' = 'latest-offset',
       'format' = 'canal-json'
   );
   create table printSink(
     id bigint,
     name string,
     description string,
     weight DECIMAL(10, 2)
      ) with (
        'connector' = 'print'
      );
   insert into printSink select * from kafkaSource;

3. 向kafka的相应topic中插入下列数据：

   {
     "data": [
       {
         "id": "111",
         "name": "scooter",
         "description": "Big 2-wheel scooter",
         "weight": "5.18"
       }
     ],
     "database": "inventory",
     "es": 1589373560000,
     "id": 9,
     "isDdl": false,
     "mysqlType": {
       "id": "INTEGER",
       "name": "VARCHAR(255)",
       "description": "VARCHAR(512)",
       "weight": "FLOAT"
     },
     "old": [
       {
         "weight": "5.15"
       }
     ],
     "pkNames": [
       "id"
     ],
     "sql": "",
     "sqlType": {
       "id": 4,
       "name": 12,
       "description": 12,
       "weight": 7
     },
     "table": "products",
     "ts": 1589373560798,
     "type": "UPDATE"
   }

4. 用户可按下述操作查看输出结果:
   • 方法一："更多" -> "FlinkUI" -> "Task Managers" -> "Stdout"。
   • 方法二：若在提交运行作业前选择了保存日志，则可以从日志的taskmanager.out文件中查看。

   -U(111,scooter,Big2-wheel scooter,5.15)
   +U(111,scooter,Big2-wheel scooter,5.18)