文档首页> 数据湖探索 DLI> Flink SQL语法参考> Flink Opensource SQL1.12语法参考> 数据定义语句DDL> Format> Debezium Format
更新时间：2023-02-24 GMT+08:00
查看PDF

Debezium Format

功能描述

Debezium是一个 CDC（Changelog Data Capture，变更数据捕获）的工具，可以把其他数据库的更改实时流式传输到 Kafka 中。 Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON消息。

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

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

参数说明

表1

参数

是否必选

默认值

是否必选

描述

format

是

(none)

String

指定要使用的格式，此处应为 '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

当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。

debezium-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' ，且会以相同的格式输出。

debezium-json.map-null-key.mode

否

'FAIL'

String

指定处理 Map 中 key 值为空的方法。 当前支持的值有FAIL、DROP和LITERAL。

  • Option 'FAIL' 将抛出异常，如果遇到 Map 中 key 值为空的数据。
  • Option 'DROP' 将丢弃 Map 中 key 值为空的数据项。
  • Option 'LITERAL' 将使用字符串常量来替换 Map 中的空 key 值。字符串常量的值由 'debezium-json.map-null-key.literal' 定义。

debezium-json.map-null-key.literal

否

'null'

String

当 'debezium-json.map-null-key.mode' 是 LITERAL 的时候，指定字符串常量替换 Map 中的空 key 值。

支持的Connector

  • Kafka

示例

使用kafka发送数据，输出到print中。

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

    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' = '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中插入下列数据：

    {
  "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. 用户可按下述操作查看输出结果:

    • 方法一："更多" -> "FlinkUI" -> "Task Managers" -> "Stdout"。
    • 方法二：若在提交运行作业前选择了保存日志，则可以从日志的taskmanager.out文件中查看。
    -U(111,scooter,Big2-wheel scooter,5.18)
+U(111,scooter,Big2-wheel scooter,5.15)

父主题： Format