0
0

37、Flink 的CDC 格式：debezium部署以及mysql示例（2）-Flink 与Debezium 实践

Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

  • 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。
  • 2、Flink基础系列 本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。
  • 3、Flik Table API和SQL基础系列 本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。
  • 4、Flik Table API和SQL提高与应用系列 本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。
  • 5、Flink 监控系列 本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

文章目录

本文详细的介绍了Flink sql client通过debezium解析cdc数据同步至kafka的使用过程。

如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。

本文除了maven依赖外，还依赖kafka、flink、debezium。

本专题文章分为如下几篇：

37、Flink 的CDC 格式：debezium部署以及mysql示例（1）-debezium的部署与示例
37、Flink 的CDC 格式：debezium部署以及mysql示例（2）-Flink 与Debezium 实践
37、Flink 的CDC 格式：debezium部署以及mysql示例（完整版）

二、Flink 与 Debezium 实践

1、maven 依赖

为了使用Debezium格式，使用构建自动化工具（如Maven或SBT）的项目和带有SQLJAR包的SQLClient都需要以下依赖项。

1）、avro

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-avro-confluent-registry</artifactId><version>1.17.1</version></dependency>

2）、json

json格式是Flink 自带的依赖包。

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>1.17.1</version></dependency>

参考 Debezium 文档，了解如何设置 Debezium Kafka Connect 用来将变更日志同步到 Kafka 主题。

2、Flink sql client 建表示例

Debezium 为变更日志提供了统一的格式，这是一个 JSON 格式的从 MySQL userscoressink表捕获的更新操作的简单示例:

{"before":{"name":"alan_test","scores":666.0},"after":{"name":"alan_test","scores":888.0},"source":{"version":"1.7.2.Final","connector":"mysql","name":"ALAN","ts_ms":1705717298000,"snapshot":"false","db":"cdctest","sequence":null,"table":"userscoressink","server_id":1,"gtid":null,"file":"alan_master_logbin.000004","pos":4931,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1705717772785,"transaction":null}

MySQL userscoressink表有2列（name，scores）。上面的 JSON 消息是 userscoressink表上的一条更新事件，其中 name = alan_test的行的 scores 值从 666 更改为 888 。

此消息已同步到 Kafka 主题 ALAN.cdctest.userscoressink，则可以使用以下 DDL 来使用此主题并解析更改事件。

具体启动debezium参考本文的第一部分的kafka示例，其他不再赘述。下面的部分仅仅是演示debezium环境都正常后，在Flink SQL client中的操作。

-- 元数据与 MySQL "userscoressink" 表完全相同CREATETABLE userscoressink_debezium (
  name STRING,
  scores FLOAT)WITH('connector'='kafka','topic'='ALAN.cdctest.userscoressink','properties.bootstrap.servers'='server1:9092,server2:9092,server3:9092','properties.group.id'='testGroup','scan.startup.mode'='earliest-offset',-- 使用 'debezium-json' format 来解析 Debezium 的 JSON 消息-- 如果 Debezium 用 Avro 编码消息，请使用 'debezium-avro-confluent''format'='debezium-json');

重要说明

上面的debezium的json更新事件是基于在配置kafka插件的时候设置了以下参数的情况下，默认该参数是true，json的事件就是下面一个示例的情况。

key.converter.schemas.enable=false 
value.converter.schemas.enable=false

在某些情况下（默认），用户在设置 Debezium Kafka Connect 时，可能会开启 Kafka 的配置 ‘value.converter.schemas.enable’，用来在消息体中包含 schema 信息。
然后，Debezium JSON 消息可能如下所示:

{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":true,"field":"name"},{"type":"double","optional":true,"field":"scores"}],"optional":true,"name":"ALAN.cdctest.userscoressink.Value","field":"before"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"name"},{"type":"double","optional":true,"field":"scores"}],"optional":true,"name":"ALAN.cdctest.userscoressink.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"ALAN.cdctest.userscoressink.Envelope"},"payload":{"before":{"name":"alan_test","scores":666.0},"after":{"name":"alan_test","scores":888.0},"source":{"version":"1.7.2.Final","connector":"mysql","name":"ALAN","ts_ms":1705717298000,"snapshot":"false","db":"cdctest","sequence":null,"table":"userscoressink","server_id":1,"gtid":null,"file":"alan_master_logbin.000004","pos":4931,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1705717772785,"transaction":null}}

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

本示例采用默认的方式进行展示。

CREATETABLE userscoressink_debezium (
  name STRING,
  scores FLOAT)WITH('connector'='kafka','topic'='ALAN.cdctest.userscoressink','properties.bootstrap.servers'='server1:9092,server2:9092,server3:9092','properties.group.id'='testGroup','scan.startup.mode'='earliest-offset','debezium-json.schema-include'='true','format'='debezium-json');

Flink SQL>CREATETABLE userscoressink_debezium (>   name STRING,>   scores FLOAT>)WITH(>'connector'='kafka',>'topic'='ALAN.cdctest.userscoressink',>'properties.bootstrap.servers'='server1:9092,server2:9092,server3:9092',>'properties.group.id'='testGroup',>'scan.startup.mode'='earliest-offset',>'debezium-json.schema-include'='true',>'format'='debezium-json'>);[INFO]Execute statement succeed.

Flink SQL>select*from userscoressink_debezium;+----+--------------------------------+--------------------------------+| op |                           name |                         scores |+----+--------------------------------+--------------------------------+|+I |                           alan |80.0||+I |                       alanchan |100.0||+I |                    alanchanchn |109.0||+I |                      alan_test |666.0||-U |                      alan_test |666.0||+U |                      alan_test |888.0||-D |                      alan_test |888.0||+I |               test_alanchanchn |199.0||-U |               test_alanchanchn |199.0||+U |               test_alanchanchn |299.0||-D |               test_alanchanchn |299.0|

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

-- MySQL "userscoressink_debezium" 的实时物化视图-- 按name分组统计scores
Flink SQL>select name ,sum(scores)from userscoressink_debezium groupby name;+----+--------------------------------+--------------------------------+| op |                           name |                         EXPR$1|+----+--------------------------------+--------------------------------+|+I |                           alan |80.0||+I |                       alanchan |100.0||+I |                    alanchanchn |109.0||+I |                      alan_test |666.0||-D |                      alan_test |666.0||+I |                      alan_test |888.0||-D |                      alan_test |888.0||+I |               test_alanchanchn |199.0||-D |               test_alanchanchn |199.0||+I |               test_alanchanchn |299.0||-D |               test_alanchanchn |299.0|

3、Available Metadata

以下格式元数据可以在表定义中公开为只读（VIRTUAL）列。

只有当相应的连接器转发格式元数据时，注意格式元数据字段才可用。截至Flink 1.17版本，只有Kafka连接器能够公开其值格式的元数据字段。

在这里插入图片描述
以下示例显示了如何访问Kafka中的Debezium元数据字段：

CREATETABLE userscoressink_debezium_meta (
  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,
  name STRING,
  scores FLOAT)WITH('connector'='kafka','topic'='ALAN.cdctest.userscoressink','properties.bootstrap.servers'='server1:9092,server2:9092,server3:9092','properties.group.id'='testGroup','scan.startup.mode'='earliest-offset','debezium-json.schema-include'='true','format'='debezium-json');

Flink SQL>CREATETABLE userscoressink_debezium_meta (>   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,>   name STRING,>   scores FLOAT>)WITH(>'connector'='kafka',>'topic'='ALAN.cdctest.userscoressink',>'properties.bootstrap.servers'='server1:9092,server2:9092,server3:9092',>'properties.group.id'='testGroup',>'scan.startup.mode'='earliest-offset',>'debezium-json.schema-include'='true',>'format'='debezium-json'>);[INFO]Execute statement succeed.

Flink SQL>select*from userscoressink_debezium_meta;+----+-------------------------+-------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+| op |               origin_ts |              event_time |                origin_database |                  origin_schema |                   origin_table |              origin_properties |                           name |                         scores |+----+-------------------------+-------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+|+I |2024-01-2002:17:48.975|2024-01-2002:17:48.972|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                           alan |80.0||+I |2024-01-2002:17:48.976|2024-01-2002:17:48.976|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                       alanchan |100.0||+I |2024-01-2002:17:48.977|2024-01-2002:17:48.976|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                    alanchanchn |109.0||+I |2024-01-2002:29:10.512|2024-01-2002:21:16.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                      alan_test |666.0||-U |2024-01-2002:29:32.785|2024-01-2002:21:38.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                      alan_test |666.0||+U |2024-01-2002:29:32.785|2024-01-2002:21:38.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                      alan_test |888.0||-D |2024-01-2002:29:56.886|2024-01-2002:22:02.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|                      alan_test |888.0||+I |2024-01-2002:53:49.248|2024-01-2002:45:55.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|               test_alanchanchn |199.0||-U |2024-01-2002:53:55.424|2024-01-2002:46:01.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|               test_alanchanchn |199.0||+U |2024-01-2002:53:55.424|2024-01-2002:46:01.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|               test_alanchanchn |299.0||-D |2024-01-2002:53:59.522|2024-01-2002:46:05.000|                        cdctest |(NULL)|                 userscoressink | {query=null, thread=null, s...|               test_alanchanchn |299.0|
Query terminated, received a total of11rows

4、Format 参数

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

1）、Debezium Avro

在这里插入图片描述

2）、Debezium Json

在这里插入图片描述

5、数据类型映射

截至Flink 1.17 版本，Debezium Format 使用 JSON Format 进行序列化和反序列化。有关数据类型映射的更多详细信息，请参考 JSON Format 文档 和 Confluent Avro Format 文档。

6、注意事项

1）、重复的变更事件

在正常的操作环境下，Debezium 应用能以 exactly-once 的语义投递每条变更事件。在这种情况下，Flink 消费 Debezium 产生的变更事件能够工作得很好。 然而，当有故障发生时，Debezium 应用只能保证 at-least-once 的投递语义。可以查看 Debezium 官方文档 了解更多关于 Debezium 的消息投递语义。 这也意味着，在非正常情况下，Debezium 可能会投递重复的变更事件到 Kafka 中，当 Flink 从 Kafka 中消费的时候就会得到重复的事件。 这可能会导致 Flink query 的运行得到错误的结果或者非预期的异常。因此，建议在这种情况下，将作业参数 table.exec.source.cdc-events-duplicate 设置成 true，并在该 source 上定义 PRIMARY KEY。 框架会生成一个额外的有状态算子，使用该 primary key 来对变更事件去重并生成一个规范化的 changelog 流。

2）、消费 Debezium Postgres Connector 产生的数据

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

当配置为 FULL 时，更新和删除事件将完整包含所有列的之前的值。当为其他配置时，更新和删除事件的 “before” 字段将只包含 primary key 字段的值，或者为 null（没有 primary key）。 你可以通过运行 ALTER TABLE REPLICA IDENTITY FULL 来更改 REPLICA IDENTITY 的配置。 请阅读 Debezium 关于 PostgreSQL REPLICA IDENTITY 的文档 了解更多。

以上，本文详细的介绍了Flink sql client通过debezium解析cdc数据同步至kafka的使用过程。

本专题文章分为如下几篇：

37、Flink 的CDC 格式：debezium部署以及mysql示例（1）-debezium的部署与示例
37、Flink 的CDC 格式：debezium部署以及mysql示例（2）-Flink 与Debezium 实践
37、Flink 的CDC 格式：debezium部署以及mysql示例（完整版）

标签： flink kafka flink hive
本文转载自: https://blog.csdn.net/chenwewi520feng/article/details/135773104
版权归原作者 一瓢一瓢的饮 alanchanchn 所有， 如有侵权，请联系我们删除。
登录后发布评论

“37、Flink 的CDC 格式：debezium部署以及mysql示例（2）-Flink 与Debezium 实践”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航