Debezium

本文为您介绍Debezium格式的使用示例、配置选项和类型映射。

背景信息

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

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

• 将增量数据从数据库同步到其他系统

• 日志审计

• 数据库的实时物化视图

• 数据库表的temporal join变更历史

Flink还支持将Flink SQL中的INSERT、UPDATE或DELETE消息编码为Debezium格式的JSON消息或Avro消息，输出到Kafka等存储中。

使用示例

假设Flink接收到MySQL products表的一条更新记录，该记录采用Debezium JSON格式，包含id、name、description和weight列信息。

{
  "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": {...},
  "op": "u",
  "ts_ms": 1589362330904,
  "transaction": null
};

上面的JSON消息是products表上的一条更新事件，其中id = 111的行的weight值从5.18更改为5.15。假设此消息已同步到一个名为products_binlog的Kafka topic，则可以使用以下DDL来使用此topic并解析更改事件。

-- 使用'debezium-json' format来解析Debezium的JSON消息
CREATE TABLE topic_products (
-- schema与MySQL的products表完全相同
  id BIGINT,
  name STRING,
  description STRING,
  weight DECIMAL(10, 2)
) WITH (
 'connector' = 'kafka',
 'topic' = 'products_binlog',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'debezium-json'
-- 使用'debezium-json' format来解析Debezium的JSON消息
-- 如果Debezium用Avro编码消息，请使用'debezium-avro-confluent'
);

在某些情况下，在设置Debezium Kafka Connect时，可能会开启Kafka的配置value.converter.schemas.enable，用来在消息体中包含schema信息。Debezium JSON消息可能如下所示：

{
  "schema": {...},
  "payload": {
    "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": {...},
    "op": "u",
    "ts_ms": 1589362330904,
    "transaction": null
  }
}

为了解析这一类信息，您需要在上述DDL WITH子句中添加选项'debezium-json.schema-include' = 'true'（默认为false）。通常情况下，建议不要包含schema的描述，因为这样会使消息变得非常冗长，并降低解析性能。

在将主题注册为Flink表之后，可以将Debezium消息用作变更日志源。

-- MySQL "products" 的实时物化视图。
-- 计算相同产品的最新平均重量。
SELECT name, AVG(weight) FROM topic_products GROUP BY name;

-- 将 MySQL "products" 表的所有数据和增量更改同步到。
-- Elasticsearch "products" 索引，供将来查找。
INSERT INTO elasticsearch_products
SELECT * FROM topic_products;

配置选项

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

debezium-avro-confluent

使用debezium-avro-confluent来解析Debezium Avro消息。

debezium-json

使用debezium-json来解析Debezium JSON消息。

类型映射

目前，Debezium使用JSON格式进行序列化和反序列化。有关数据类型映射的更多详细信息，请参考JSON Format文档和Confluent Avro Format文档。

其他使用说明

可用的元数据

以下格式元数据可以在DDL语句中声明为只读（VIRTUAL）列。

下面的例子展示了如何在Kafka中访问Debezium元数据字段：

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' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
);

常见问题

故障时投递重复的变更事件

在正常的操作环境下，Debezium能够以exactly-once的语义投递每条变更事件，Flink能够正常消费Debezium产生的变更事件。在非正常情况下（例如有故障发生），Debezium只能保证at-least-once的投递语义。此时，Debezium可能会投递重复的变更事件到Kafka中，当Flink从Kafka中消费的时候就会得到重复的事件，可能导致Flink query的运行得到错误的结果或者非预期的异常。因此，在这种情况下，建议将作业参数table.exec.source.cdc-events-duplicate设置成true，并在该源表上定义PRIMARY KEY。Flink系统会生成一个额外的有状态算子，使用该PRIMARY KEY来对变更事件去重并生成一个规范化的changelog流。

无法正确解析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的配置。