更新时间:2023-04-13 GMT+08:00

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)