kettle从入门到精通第六十九课 ETL之kettle kettle cdc mysql，轻松实现实时增量同步

kettle,etl,cdc,mysql · 浏览次数 : 0

小编点评

本文介绍了如何使用Kettle的插件Debezium直接从MySQL捕获变更数据（CDC）。首先，简要介绍了CDC技术及其在MySQL中的应用。接着，详细描述了Debezium的四个主要步骤：解析MySQL二进制日志文件、解析JSON字符串、根据操作类型路由事件以及编写日志步骤以模拟后续操作。然后，展示了如何配置Debezium步骤以仅输出数据修改（DML）操作，并设置了DDL（数据定义语言）输出。最后，给出了一个示例，其中包含了创建、更新和删除操作的数据字典定义。 1. **CDC技术及Debezium概述**：本文首先介绍了CDC的技术概念，它是一种用于捕获和同步数据库中更改的技术。接着，提出了使用Debezium直接从MySQL捕获变更数据的方案。 2. **Debezium步骤解析MySQL二进制日志**：详细阐述了Debezium的四个主要步骤： - 解析MySQL二进制日志文件。 - 解析JSON字符串。 - 根据操作类型（如CREATE、UPDATE、DELETE等）进行路由。 - 写入日志步骤以模拟后续的操作。 3. **配置步骤**：说明了Debezium步骤的配置方法，包括： - 设置输出为DML（数据修改）。 - 输出DDL（数据定义语言）。 - 配置其他相关属性。 4. **示例与数据字典**：通过一个具体的示例，展示了binlog解析后的DML和DDL语句的结构。这个示例包括了创建、更新和删除操作的数据字典定义。总的来说，本文全面地介绍了如何使用Debezium和Kettle插件来捕获和同步MySQL数据库中的变更数据，包括其配置、操作步骤以及示例。

正文

1、之前kettle cdc mysql的时候使用的方案是canal+kafka+kettle，今天我们一起学习下使用kettle的插件Debezium直接cdc mysql。

　注：CDC (Change Data Capture) 是一种技术，用于捕获和同步数据库中的更改。

1）Debezium步骤解析mysql binlog日志。

2）json input步骤解析json字符串。

3）switch-case 根据op字段进行路由。

4）create、delete、update、ddl是写日志步骤，模拟后续操作。

2、Debezium步骤配置如下（只输出dml），更多属性配置参考Debezium官网。如下图所示：

3、switch-case配置如下，不做过多介绍。

4、binlog解析之后的dml语句对应的json字符串结构

CREATE

{
    "ts_ms": 1718273338000,
    "db": "test",
    "table": "t1",
    "op": "CREATE",
    "after": "{\"id\":6,\"name\":\"金刚葫芦娃\",\"createtime\":1718297698000}",
    "pk": "{\"id\":6}"
}

UPDATE

{
    "ts_ms": 1718273345000,
    "db": "test",
    "table": "t1",
    "op": "UPDATE",
    "before": "{\"id\":6,\"name\":\"金刚葫芦娃\",\"createtime\":1718297698000}",
    "after": "{\"id\":6,\"name\":\"金刚葫芦娃plus\",\"createtime\":1718297698000}",
    "pk": "{\"id\":6}"
}

DELETE

{
    "ts_ms": 1718273369000,
    "db": "test",
    "table": "t1",
    "op": "DELETE",
    "before": "{\"id\":6,\"name\":\"金刚葫芦娃plus\",\"createtime\":1718297698000}",
    "pk": "{\"id\":6}"
}

5、Debezium步骤配置中设置ddl为true（输出dml和ddl），更多属性配置参考Debezium官网。如下图所示：

6、binlog解析之后的ddl语句对应的json字符串结构

DROP TABLE

{
    "ts_ms": 1718274150072,
    "db": "test",
    "table": "t3",
    "op": "ddl",
    "sql": "DROP TABLE `test`.`t3`"
}

CREATE TABLE

{
    "ts_ms": 1718274165789,
    "db": "test",
    "table": "t3",
    "op": "ddl",
    "ddl_type": "CREATE",
    "sql": "CREATE TABLE `t3` (\n `id` bigint NOT NULL AUTO_INCREMENT,\n `name` varchar(10) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,\n `createtime` datetime DEFAULT NULL,\n PRIMARY KEY (`id`)\n ) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci"
}