NIFI实现数据库数据增量同步

说明

nifi版本：1.23.2（docker镜像）

需求背景

将数据库中的数据同步到另一个数据库中，要求对于新增的数据和历史有修改的数据进行增量同步

模拟数据

建表语句

源数据库和目标数据库结构要保持一致，这样可以避免后面单独转换

-- 创建测试表
CREATE TABLE `sys_user` (`id` bigint NOT NULL AUTO_INCREMENT COMMENT '用户ID',`name` varchar(50) NOT NULL DEFAULT '' COMMENT '姓名',`age`  int NOT NULL DEFAULT 0 COMMENT '年龄',`gender` tinyint NOT NULL COMMENT '性别,1:男,0:女',`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`modify_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',`is_deleted` tinyint NOT NULL DEFAULT '0' COMMENT '是否已删除',PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT  CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci ROW_FORMAT=DYNAMIC COMMENT='用户表';

测试数据

-- 模拟数据
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据1', 20, 1);
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据2', 21, 1);
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据3', 21, 0);
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据4', 18, 0);
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据5', 22, 1);

完整测试数据

配置数据库连接池

在画布空白位置鼠标右键，选择Configure

新增配置

在弹出的界面点击+号，添加新的数据库连接池配置，如果已经有了配置该步骤可以跳过

在弹出的界面筛选对应类型的连接池，我这里选择DBCPConnectionPool，然后点击ADD

点击刚才新添加的那一条数据右侧的小齿轮，进行连接池相关的配置

配置连接池相关属性

主要配置以下几个内容，其他的根据情况决定是否需要修改,密码输入后是不会显示的

校验属性

校验配置是否正确，点击右上角的对钩，然后在弹出的界面点击VERIFY进行验证

验证通过会全部显示绿色，如果某一条不通过会有提示，最后点击APPLY

（可选操作）给配置起个名字

为了方便后续使用，给连接池起个名字，要不然以后配置多了会分不清

激活连接池的配置

点击右侧的闪电标志激活配置，在新的页面中点击ENABLE激活，最后点击CLOSE关闭

已激活的配置

同理增加目标数据库的连接池配置，步骤和上面是一样的这里不再重复了，最终配置好后会有两个连接池的配置。如下：

获取数据库表数据

添加处理器：QueryDatabaseTable

点击工具栏的Processor，拖拽到画布中，筛选QueryDatabaseTable处理器，然后点击ADD添加到画布中

配置处理器：QueryDatabaseTable

双击处理器，切换到PROPERTIES选项卡,配置以下内容

Maximum-value Columns（最大值列）：官方文档是这么解释的：以逗号分隔的列名列表。处理器将跟踪自处理器开始运行以来返回的每一列的最大值。使用多个列意味着列列表的顺序，并且每列的值预计比前几列的值增加得更慢。因此，使用多个列意味着列的分层结构，通常用于对表进行分区。此处理器可用于仅检索自上次检索以来添加/更新的那些行。请注意，某些 JDBC 类型（如 bit/boolean）不利于保持最大值，因此这些类型的列不应列在此属性中，并且将导致处理过程中的错误。如果未提供列，则将考虑表中的所有行，这可能会对性能产生影响。注意：为给定表使用一致的最大值列名非常重要，这样增量提取才能正常工作。
支持表达式语言：true

校验属性

给处理器起个名字，表示当前整个工作流的作用

拆分数据

添加处理器：SplitAvro

配置处理器：SplitAvro

双击处理器，切换到PROPERTIES选项卡,所有内容默认即可

数据入库

添加处理器：PutDatabaseRecord

配置处理器：PutDatabaseRecord

双击处理器，切换到PROPERTIES选项卡

新增Record Reader

配置AvroReader

点击右侧的箭头，在弹出的界面选择刚才配置的Reader,然后点击右侧的小齿轮

在弹出的界面根据自己的需要自行配置，这里按照默认的配置即可

激活Reader

点击右侧的闪电标志进行激活

激活后的状态变为Enabled

其他配置

校验属性

连接所有处理器

连接处理器

连接QueryDatabaseTable和SplitAvro两个处理器，勾选For Relationships下的success

连接SplitAvro和PutDatabaseRecord两个处理器，勾选For Relationships下的split

处理SplitAvro处理器的告警

双击SplitAvro处理器，切换到RELATIONSHIPS，勾选下面的两个选项，然后点击APPLY

处理PutDatabaseRecord处理器的告警

双击PutDatabaseRecord处理器，切换到RELATIONSHIPS，勾选下面的选项，然后点击APPLY

完整配置

启动所有处理器

QueryDatabaseTable处理器默认是一分钟执行一次的，可以在SCHEDULING选项卡下面进行配置，这里按照默认的时间来执行

在画布空白位置鼠标右键选择Start启动所有的处理器

查看目标数据库数据

等待一分钟后查看目标数据库数据，发现源数据库的5条数据被同步到了目标数据库

修改源数据库的数据

UPDATE sys_user SET is_deleted = 1 WHERE id = 1;
UPDATE sys_user SET is_deleted = 1 WHERE id = 4;
INSERT INTO sys_user (name, age, gender) VALUES ('测试数据6', 22, 1);