格式, 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 SQL暂不支持直接处理同时包含了UPDATE_BEFORE和UPDATE_AFTER的单个Update类型消息,因此Flink在写入数据时,UPDATE_BEFORE和UPDATE_AFTER类型的数据将被分别转换为DELETE和INSERT类型,并使用Debezium编码格式写入到下游存储。
使用示例
假设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
};
示例中各字段的含义,详情请参见Debezium。
上面的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消息。
参数 | 是否必选 | 默认值 | 类型 | 描述 |
format | 是 | (none) | String | 指定要使用的格式,解析Debezium Avro消息时,参数取值为debezium-avro-confluent。 |
debezium-avro-confluent.basic-auth.credentials-source | 否 | (none) | String | Schema Registry的基本身份验证凭据的source。 |
debezium-avro-confluent.basic-auth.user-info | 否 | (none) | String | Schema Registry的基本认证用户信息。 |
debezium-avro-confluent.bearer-auth.credentials-source | 否 | (none) | String | Schema Registry的持有者身份验证凭据的source。 |
debezium-avro-confluent.bearer-auth.token | 否 | (none) | String | Schema Registry的持有者认证令牌。 |
debezium-avro-confluent.properties | 否 | (none) | Map | 属性映射,该映射被转发到Schema Registry。对于没有通过Flink配置选项正式公开的选项非常有用。 重要 Flink选项具有更高的优先级。 |
debezium-avro-confluent.ssl.keystore.location | 否 | (none) | String | SSL keystore的位置。 |
debezium-avro-confluent.ssl.keystore.password | 否 | (none) | String | SSL keystore的密码。 |
debezium-avro-confluent.ssl.truststore.location | 否 | (none) | String | SSL truststore的位置。 |
debezium-avro-confluent.ssl.truststore.password | 否 | (none) | String | SSL truststore的密码。 |
debezium-avro-confluent.subject | 否 | (none) | String | Confluent Schema Registry subject,在该subject下注册此格式在序列化期间使用的模式。默认情况下,如果使用kafka和upsert-kafka连接器作为值或键格式,则使用<Topname>-value或<topname>-key作为默认subject名称。对于filesystem连接器,当其作为sink使用时,必须使用subject选项。 |
debezium-avro-confluent.url | 是 | (none) | String | 用于获取或注册schemas的Confluent Schema Registry的URL。 |
debezium-json
使用debezium-json来解析Debezium JSON消息。
参数 | 是否必选 | 默认值 | 类型 | 描述 |
format | 是 | (none) | String | 指定要使用的格式,解析Debezium JSON消息时,参数取值为debezium-json。 |
debezium-json.schema-include | 否 | false | Boolean | 设置Debezium Kafka Connect时,可以启用Kafka配置value.converter.schemas.enable,以在消息中包含schema。此选项表明Debezium JSON消息是否包含schema。 参考取值如下:
|
debezium-json.ignore-parse-errors | 否 | false | Boolean | 参数取值如下:
|
debezium-json.timestamp-format.standard | 否 | SQL | String | 指定输入和输出时间戳格式。参数取值如下:
|
debezium-json.map-null-key.mode | 否 | FAIL | String | 指定处理Map中key值为空的方法。参数取值如下:
|
debezium-json.map-null-key.literal | 否 | null | String | 当debezium-json.map-null-key.mode的值是LITERAL时,指定字符串常量替换Map中的空key值。 |
debezium-json.encode.decimal-as-plain-number | 否 | false | Boolean | 参数取值如下:
|
类型映射
目前,Debezium使用JSON格式进行序列化和反序列化。有关数据类型映射的更多详细信息,请参考JSON Format文档和Confluent Avro Format文档。
其他使用说明
可用的元数据
以下格式元数据可以在DDL语句中声明为只读(VIRTUAL)列。
只有当相应的连接器转发格式元数据时,格式元数据字段才可用。目前,只有Kafka连接器能够声明其值格式的元数据字段。
键 | 数据类型 | 说明 |
schema | STRING NULL | 描述有效负载模式的JSON字符串。如果模式未包含在Debezium记录中,则为Null。 |
ingestion-timestamp | TIMESTAMP_LTZ(3) NULL | 连接器处理事件时的时间戳。对应于Debezium记录中的ts_ms字段。 |
source.timestamp | TIMESTAMP_LTZ(3) NULL | 源系统创建事件的时间戳。对应于Debezium记录中的source.ts_ts字段。 |
source.database | STRING NULL | 原始数据库。对应于Debezium记录中的source.db字段(如果可用)。 |
source.schema | STRING NULL | 原始数据库的模式。对应于Debezium记录中的source.schema字段(如果可用)。 |
source.table | STRING NULL | 原始数据库的表。对应于Debezium记录中的source.table或source.collection字段(如果可用)。 |
source.properties | MAP<STRING, STRING> NULL | 各种源属性的映射。对应于Debezium记录中的source字段。 |
下面的例子展示了如何在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的消息投递语义的更多信息,请参见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的配置。
更多信息,请参见Debezium connector for PostgreSQL。