游仙区专业网站建设价格,php wordpress面试题,注册小公司流程和费用,公司网站制作网络公司Flink系列之#xff1a;Table API Connectors之Debezium 一、Debezium二、依赖三、使用Debezium Format四、可用元数据五、Format参数六、重复的变更事件七、消费 Debezium Postgres Connector 产生的数据八、数据类型映射 一、Debezium
Debezium 是一个 CDC#xff08;Chan… Flink系列之Table API Connectors之Debezium 一、Debezium二、依赖三、使用Debezium Format四、可用元数据五、Format参数六、重复的变更事件七、消费 Debezium Postgres Connector 产生的数据八、数据类型映射 一、Debezium
Debezium 是一个 CDCChangelog 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 Avro
为了使用 Debezium 格式使用构建自动化工具例如 Maven 或 SBT的项目和带有 SQL JAR 包的 SQL 客户端都需要以下依赖项。
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-avro-confluent-registry/artifactIdversion1.18.0/version
/dependencyDebezium Json
为了使用 Debezium 格式使用构建自动化工具例如 Maven 或 SBT的项目和带有 SQL JAR 包的 SQL 客户端都需要以下依赖项。
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-json/artifactIdversion1.18.0/version
/dependency
三、使用Debezium Format
Debezium 为变更日志提供了统一的格式这是一个 JSON 格式的从 MySQL product 表捕获的更新操作的简单示例:
{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
}MySQL 产品表有4列id、name、description、weight。上面的 JSON 消息是 products 表上的一条更新事件其中 id 111 的行的 weight 值从 5.18 更改为 5.15。假设此消息已同步到 Kafka 主题 products_binlog则可以使用以下 DDL 来使用此主题并解析更改事件。
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,-- 使用 debezium-json format 来解析 Debezium 的 JSON 消息-- 如果 Debezium 用 Avro 编码消息请使用 debezium-avro-confluentformat 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;四、可用元数据
以下格式元数据可以在表定义中公开为只读虚拟列。
注意 仅当相应的连接器转发格式元数据时格式元数据字段才可用。目前只有 Kafka 连接器能够公开其值格式的元数据字段。
KeyData TypeDescriptionschemaSTRING NULL描述负载模式的 JSON 字符串。如果架构未包含在 Debezium 记录中则为 Null。ingestion-timestampTIMESTAMP_LTZ(3) NULL连接器处理事件的时间戳。对应于 Debezium 记录中的 ts_ms 字段。source.timestampTIMESTAMP_LTZ(3) NULL源系统创建事件的时间戳。对应于 Debezium 记录中的 source.ts_ms 字段。source.databaseSTRING NULL原始数据库。对应于 Debezium 记录中的 source.db 字段如果可用。source.schemaSTRING NULL原始数据库架构。对应于 Debezium 记录中的 source.schema 字段如果可用。source.tableSTRING NULL原始数据库表。对应于 Debezium 记录中的 source.table 或 source.collection 字段如果可用。source.propertiesMAPSTRING, STRING NULL各种源属性的映射。对应于 Debezium 记录中的源字段。
以下示例展示了如何访问 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 MAPSTRING, 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
);五、Format参数
Flink 提供了 debezium-avro-confluent 和 debezium-json 两种 format 来解析 Debezium 生成的 JSON 格式和 Avro 格式的消息。 请使用 debezium-avro-confluent 来解析 Debezium 的 Avro 消息使用 debezium-json 来解析 Debezium 的 JSON 消息。
Debezium Avro
参数是否必选默认值类型描述formatrequired(none)String指定使用什么格式这里应该是“debezium-avro-confluence”。debezium-avro-confluent.basic-auth.credentials-sourceoptional(none)String架构注册表的基本身份验证凭据源debezium-avro-confluent.basic-auth.user-infooptional(none)String架构注册表的基本身份验证用户信息debezium-avro-confluent.bearer-auth.credentials-sourceoptional(none)String架构注册表的承载身份验证凭据源debezium-avro-confluent.bearer-auth.tokenoptional(none)String架构注册表的承载身份验证令牌debezium-avro-confluent.propertiesoptional(none)Map转发到底层架构注册表的属性映射。这对于未通过 Flink 配置选项正式公开的选项很有用。但请注意Flink 选项具有更高的优先级。debezium-avro-confluent.ssl.keystore.locationoptional(none)StringSSL 密钥库的位置/文件debezium-avro-confluent.ssl.keystore.passwordoptional(none)StringSSL 密钥库的密码debezium-avro-confluent.ssl.truststore.locationoptional(none)Stringdebezium-avro-confluent.ssl.truststore.passwordoptional(none)StringSSL 信任库的密码debezium-avro-confluent.schemaoptional(none)String已在 Confluence 模式注册表中注册或将要注册的模式。如果没有提供 schemaFlink 会将表 schema 转换为 avro schema。提供的模式必须与 Debezium 模式匹配Debezium 模式是可以为空的记录类型包括字段“before”、“after”、“op”。debezium-avro-confluent.subjectoptional(none)String融合模式注册表主题在序列化期间注册此格式使用的模式。默认情况下如果将此格式用作值或键格式则“kafka”和“upsert-kafka”连接器使用“topic_name-value”或“topic_name-key”作为默认主题名称。但对于其他连接器例如“文件系统”用作接收器时需要主题选项。debezium-avro-confluent.urlrequired(none)String用于获取/注册模式的 Confluence 模式注册表的 URL。
Debezium Json
参数是否必选默认值类型描述format必选(none)String指定要使用的格式此处应为 ‘debezium-json’。debezium-json.schema-include可选falseBoolean设置 Debezium Kafka Connect 时用户可以启用 Kafka 配置 ‘value.converter.schemas.enable’ 以在消息中包含 schema。此选项表明 Debezium JSON 消息是否包含 schema。debezium-json.ignore-parse-errors可选falseBoolean当解析异常时是跳过当前字段或行还是抛出错误失败默认为 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 值。debezium-json.encode.decimal-as-plain-number选填falseBoolean将所有 DECIMAL 类型的数据保持原状不使用科学计数法表示。例0.000000027 默认会表示为 2.7E-8。当此选项设为 true 时则会表示为 0.000000027。
六、重复的变更事件
在正常的操作环境下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 流。
“table.exec.source.cdc-events-duplicate” 是 Flink 的一个配置选项用于处理 Change Data Capture (CDC) 事件的重复数据。CDC 是一种常见的数据库技术用于捕获数据库中的数据更改并将其作为事件流进行处理。
在 Flink 中“table.exec.source.cdc-events-duplicate” 选项用于指定如何处理 CDC 事件流中的重复数据。具体来说它可以接受以下两个值 “strict”严格模式。如果启用了严格模式Flink 会严格检查事件流中是否存在重复的 CDC 事件。如果检测到重复事件出现Flink 会抛出一个异常并停止任务执行。 “lenient”宽松模式。如果启用了宽松模式Flink 会使用一个简单的算法来检测重复事件并尽量过滤掉这些重复事件。但是它不能保证完全消除重复事件的可能性。
使用时可以在 Flink 的配置文件或在代码中通过设置相应的属性来指定 “table.exec.source.cdc-events-duplicate” 的值。例如在 Flink 的配置文件中可以添加以下行来启用宽松模式
table.exec.source.cdc-events-duplicate: lenient七、消费 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 Format 使用 JSON Format 进行序列化和反序列化。
当前JSON schema 将会自动从 table schema 之中自动推导得到。不支持显式地定义 JSON schema。
在 Flink 中JSON Format 使用 jackson databind API 去解析和生成 JSON。
下表列出了 Flink 中的数据类型与 JSON 中的数据类型的映射关系。
Flink SQL类型JSON类型CHAR/VARCHAR/STRINGstringBOOLEANbooleanBINARY/VARBINARYstring with encoding: base64DECIMALnumberTINYINTnumberSMALLINTnumberINTnumberBIGINTnumberFLOATnumberDOUBLEnumberDATEstring with format: dateTIMEstring with format: timeTIMESTAMPstring with format: date-timeTIMESTAMP_WITH_LOCAL_TIME_ZONEstring with format: date-time (with UTC time zone)INTERVALnumberARRAYarrayMAP / MULTISETobjectROWobject
