Canal是一个CDC（ChangeLog Data Capture，变更日志数据捕获）工具，可以实时地将MySQL变更传输到其他系统。Canal为变更日志提供了统一的数据格式，并支持使用JSON或protobuf序列化消息（Canal默认使用protobuf）。支持Canal格式的连接器有消息队列Kafka和对象存储OSS。
Flink支持将Canal的JSON消息解析为INSERT、UPDATE或DELETE消息到Flink SQL系统中。在很多情况下，利用Canal这个特性非常的有用，例如：
将增量数据从数据库同步到其他系统
日志审计
数据库的实时物化视图
数据库表的temporal join变更历史
Flink还支持将Flink SQL中的INSERT、UPDATE或DELETE消息编码为Canal格式的JSON消息，输出到Kafka等存储中。
重要
目前Flink还不支持将UPDATE_BEFORE和UPDATE_AFTER合并为一条UPDATE消息。因此，Flink将UPDATE_BEFORE和UPDATE_AFTER分别编码为DELETE和INSERT类型的Canal消息。
将Kafka topic注册成Flink表之后，您可以将Canal消息用作变更日志源。
-- 关于MySQL "products" 表的实时物化视图。
-- 计算相同产品的最新平均重量。
SELECT name, AVG(weight) FROM topic_products GROUP BY name;
-- 将MySQL "products" 表的所有数据和增量更改同步到Elasticsearch "products" 索引以供将来搜索。
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;| 选项 | 要求 | 默认 | 类型 | 描述 |
| format | 必填 | (none) | String | 指定要使用的格式，使用Canal格式时，参数取值为canal-json。 |
| canal-json.ignore-parse-errors | 选填 | false | Boolean | 参数取值如下：
|
| canal-json.timestamp-format.standard | 选填 | SQL | String | 指定输入和输出时间戳格式。参数取值如下：
|
| canal-json.map-null-key.mode | 选填 | FAIL | String | 指定处理Map中key值为空的方法。参数取值如下：
|
| canal-json.map-null-key.literal | 选填 | null | String | 当canal-json.map-null-key.mode的值是LITERAL时，指定字符串常量替换Map中的空key值。 |
| canal-json.encode.decimal-as-plain-number | 选填 | false | Boolean | 参数取值如下：
|
| canal-json.database.include | 选填 | (none) | String | 一个可选的正则表达式，通过正则匹配Canal记录中的database元字段，仅读取指定数据库的changelog记录。正则字符串与Java的Pattern兼容。 |
| canal-json.table.include | 选填 | (none) | String | 一个可选的正则表达式，通过正则匹配Canal记录中的table元字段，仅读取指定表的changelog记录。正则字符串与Java的Pattern兼容。 |
目前，Canal使用JSON格式进行序列化和反序列化。有关数据类型映射的更多详细信息，请参阅JSON Format。Canal格式额外兼容了数据传输服务DTS在Kafka集群存储使用的Canal扩展变更类型（INIT）。请参见Kafka集群的数据存储格式。
下面的格式元数据可以在DDL语句中声明为只读（VIRTUAL）列。
重要
格式元数据字段只有在相应的连接器转发格式元数据时才可用。目前，只有Kafka连接器能够声明其值格式的元数据字段。
| 键 | 数据类型 | 说明 |
| database | STRING NULL | 原始数据库。对应于Canal记录中的database字段。 |
| table | STRING NULL | 原始数据库的表。对应于Canal记录中的table字段。 |
| sql-type | MAP<STRING, INT> NULL | 各种sql类型的映射。对应于Canal记录中的sqlType字段。 |
| pk-names | ARRAY<STRING> NULL | 主键名称数组。对应于Canal记录中的pkNames字段。 |
| ingestion-timestamp | TIMESTAMP_LTZ(3) NULL | 连接器处理事件时的时间戳。对应于Canal记录中的ts字段。 |
如何在Kafka中访问Canal元数据字段的代码示例如下。
CREATE TABLE KafkaTable (
origin_database STRING METADATA FROM 'value.database' VIRTUAL,
origin_table STRING METADATA FROM 'value.table' VIRTUAL,
origin_sql_type MAP<STRING, INT> METADATA FROM 'value.sql-type' VIRTUAL,
origin_pk_names ARRAY<STRING> METADATA FROM 'value.pk-names' VIRTUAL,
origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,
user_id BIGINT,
item_id BIGINT,
behavior STRING
) WITH (
'connector' = 'kafka',
'topic' = 'user_behavior',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'earliest-offset',
'value.format' = 'canal-json'
);在正常的操作环境下，Canal能够以exactly-once的语义投递每条变更事件，Flink能够正常消费Canal产生的变更事件。在非正常情况下（例如有故障发生），Canal只能保证at-least-once的投递语义。此时，Canal可能会投递重复的变更事件到Kafka中，当Flink从Kafka中消费的时候就会得到重复的事件，可能导致Flink query的运行得到错误的结果或者非预期的异常。因此，在这种情况下，建议将作业参数table.exec.source.cdc-events-duplicate设置成true，并在该source上定义PRIMARY KEY。Flink系统会生成一个额外的有状态算子，使用该PRIMARY KEY来对变更事件去重并生成一个规范化的changelog流。?
?