# 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 消息。
#### 参数说明
表1参数说明 
| 参数                                   | 是否必选 | 默认值                                          | 类型                                         | 说明                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
|:---|:---|:---|:---|:---|
| format                               | 是    | (none)                                       | String                                     | 指定要使用的格式，此处应为 'canal-json'.                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
| canal-json.ignore-parse-errors       | 否    | false                                        | Boolean                                    | 当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。                                                                                                                                                                                                                                                                                                                                                                                                                            |
| canal-json.timestamp-format.standard | 否    | 'SQL'                                        | String                                     | 指定输入和输出时间戳格式。当前支持的值是：'SQL'和'ISO-8601'。 - 选项 'SQL' 将解析 "yyyy-MM-dd HH:mm:ss.s{precision}" 格式的输入时间戳，例如 '2020-12-30 12:13:14.123'，并以相同格式输出时间戳。  - 选项 'ISO-8601' 将解析 "yyyy-MM-ddTHH:mm:ss.s{precision}" 格式的输入时间戳，例如 '2020-12-30T12:13:14.123'，并以相同的格式输出时间戳。                                              |
| canal-json.map-null-key.mode         | 否    | 'FALL'                                       | String                                     | 指定处理 Map 中 key 值为空的方法. 当前支持的值有'FAIL', 'DROP'和 'LITERAL'。 - Option 'FAIL' 将抛出异常，如果遇到 Map 中 key 值为空的数据。  - Option 'DROP' 将丢弃 Map 中 key 值为空的数据项。  - Option 'LITERAL' 将使用字符串常量来替换 Map 中的空 key 值。字符串常量的值由 'canal-json.map-null-key.literal' 定义。   |
| canal-json.map-null-key.literal      | 否    | 'null'                                       | String                                     | 当 'canal-json.map-null-key.mode' 是 LITERAL 的时候，指定字符串常量替换 Map 中的空 key 值。                                                                                                                                                                                                                                                                                                                                                                                                                             |
| canal-json.database.include          | 否    | (none)  | String  | 仅读取指定数据库的 changelog 记录（通过对比 Canal 记录中的 "database" 元数据字段）。                                                                                                                                                                                                                                                                                                                                                                                                                                           |
| canal-json.table.include             | 否    | (none)                                       | String                                     | 仅读取指定表的 changelog 记录（通过对比 Canal 记录中的 "table" 元数据字段）。                                                                                                                                                                                                                                                                                                                                                                                                                                                |
   
#### 支持的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)
   ```