Debezium Format

功能描述

Debezium是一个 CDC（Changelog Data Capture，变更数据捕获）的工具，可以把来自 MySQL、PostgreSQL、Oracle、Microsoft SQL Server 和许多其他数据库的更改实时流式传输到 Kafka 中。 Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON 和 Apache Avro 序列化消息。

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

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

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

更多具体使用可参考开源社区文档：Debezium Format。

支持的Connector

• Kafka
• Filesystem

注意事项

• 重复的变更事件

  在正常的操作环境下，Debezium 应用能以exactly-once的语义投递每条变更事件。在这种情况下，Flink 消费 Debezium 产生的变更事件能够工作得很好。 单当发生故障时，Debezium应用只能保证at-least-once的投递语义。即在非正常情况下，Debezium可能会投递重复的变更事件到Kafka中，当Flink从Kafka中消费的时候就会得到重复的事件。 这可能会导致Flink query的运行得到错误的结果或者非预期的异常。

  解决方案：将作业参数 table.exec.source.cdc-events-duplicate 设置成true，并在该source上定义PRIMARY KEY。

  框架会生成一个额外的有状态算子，使用该primary key来对变更事件去重并生成一个规范化的changelog流。

  更新信息请参考Debezium 官方文档。

• 消费Debezium Postgres Connector产生的数据

  如果你正在使用Debezium PostgreSQL Connector捕获变更到 Kafka，请确保被监控表的REPLICA IDENTITY 已经被配置成FULL ，默认值是DEFAULT。 否则，Flink SQL将无法正确解析Debezium数据。

  当配置为 FULL 时，更新和删除事件将完整包含所有列的之前的值。

  当为其他配置时，更新和删除事件的“before”字段将只包含primary key字段的值，或者为 null（没有 primary key）。

  您可以通过运行 ALTER TABLE <your-table-name> REPLICA IDENTITY FULL 来更改 REPLICA IDENTITY 的配置。

参数说明

Flink 提供了 debezium-avro-confluent 和 debezium-json 两种 format 来解析 Debezium 生成的 JSON 格式和 Avro 格式的消息。 请使用 debezium-avro-confluent 来解析 Debezium 的 Avro 消息，使用 debezium-json 来解析 Debezium 的 JSON 消息。

元数据

元数据可以在 DDL 中作为只读（虚拟）meta 列声明。

Format的元数据只有在相应的连接器使用元数据时才可用。目前，只有Kafka连接器。

元数据的使用用例参考如下：

CREATE TABLE KafkaTable (
  origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,
  event_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
  origin_database STRING METADATA FROM 'value.source.database' VIRTUAL,
  origin_schema STRING METADATA FROM 'value.source.schema' VIRTUAL,
  origin_table STRING METADATA FROM 'value.source.table' VIRTUAL,
  origin_properties MAP<STRING, STRING> METADATA FROM 'value.source.properties' VIRTUAL,
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafkaTopic',
  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
  'properties.group.id' = 'GroupId',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
);

示例

使用kafka解析Debezium Json数据，并将结果输出到print中。

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

   CREATE TABLE kafkaSource (
     id bigint,
     name string,
     description string,  
     weight DECIMAL(10, 2)
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'kafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'debezium-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中插入下列数据，该数据表示MySQL 产品表有4列（id、name、description、weight）。该JSON 消息是 products 表上的一条更新事件，其中 id = 111 的行的 weight 值从 5.18 更改为 5.15。

   {
     "before": {
       "id": 111,
       "name": "scooter",
       "description": "Big 2-wheel scooter",
       "weight": 5.18
     },
     "after": {
       "id": 111,
       "name": "scooter",
       "description": "Big 2-wheel scooter",
       "weight": 5.15
     },
     "source": {
       "version": "0.9.5.Final",
   	"connector": "mysql",
   	"name": "fullfillment",
   	"server_id" :1,
   	"ts_sec": 1629607909,
   	"gtid": "mysql-bin.000001",
   	"pos": 2238,"row": 0,
   	"snapshot": false,
   	"thread": 7,
   	"db": "inventory",
   	"table": "test",
   	"query": null},
     "op": "u",
     "ts_ms": 1589362330904,
     "transaction": null
   }

4. 按照如下方式查看taskmanager.out文件中的数据结果：
   1. 登录DLI管理控制台，选择“作业管理 > Flink作业”。
   2. 单击对应的Flink作业名称，选择“运行日志”，单击“OBS桶”，根据作业运行的日期，找到对应日志的文件夹。
   3. 进入对应日期的文件夹后，找到名字中包含“taskmanager”的文件夹进入，下载获取.out文件查看结果日志。

   -U[111, scooter, Big 2-wheel scooter, 5.18]
   +U[111, scooter, Big 2-wheel scooter, 5.15]