Kettle 实现数据增量同步方案

作业结构示例

1. 获取时间戳转换 (trans_get_timestamp)

此步骤负责读取上次的同步点。

SELECT DATE_FORMAT(time_stamp, '%Y-%m-%d %H:%i:%S') time_stamp 
FROM sync_timestamp 
WHERE table_name = 'sync_t1'

随后将结果设置为变量 ${TIME_STAMP}，供后续步骤调用。

2. 数据同步转换 (trans_sync_data)

核心逻辑在此处完成。

表输入：筛选增量数据

SELECT * FROM sync_t1 
WHERE update_time >= TO_DATE('${TIME_STAMP}', 'yyyy-mm-dd hh24:mi:ss')

插入/更新：根据主键匹配，若存在则更新，否则插入。

调度策略优化

Kettle 的 kitchen 命令行工具每次启动都会初始化数百兆内存，耗时约 10 秒以上，且一个进程只能执行一次任务。如果高并发启动大量进程，会对服务器内存造成巨大压力。

为此，我们将不同频率的任务打包成批处理文件，控制并发量。例如每分钟执行一次的作业，可以拆分为三个文件，每个文件串行执行 3 个任务：

• job_1min_1.bat：包含 job1-job3
• job_1min_2.bat：包含 job4-job6
• job_1min_3.bat：包含 job7-job9

批处理脚本示例：

e:
cd e:\pdi\data-integration
kitchen /file:e:\mykettle\job1\job.kjb /level:Error
kitchen /file:e:\mykettle\job2\job.kjb /level:Error
kitchen /file:e:\mykettle\job3\job.kjb /level:Error

*注意：Kettle 自身没有防止并发执行的机制。如果任务耗时超过调度间隔，可能会发生重叠。此时需监控日志，发现重复数据报错时及时调整调度策略。

监控与告警

为了及时发现异常，我们建立了日志表并对接 Zabbix 监控。

1. 创建日志表

CREATE TABLE `sync_log` (
  `ID_JOB` bigint(20) NOT NULL,
  `JOBNAME` varchar(100) COLLATE utf8mb4_bin DEFAULT NULL,
  `STATUS` varchar(20) COLLATE utf8mb4_bin DEFAULT NULL,
  `ERRORS` bigint(20) DEFAULT NULL,
  `LOGDATE` datetime DEFAULT NULL,
  `STARTDATE` datetime DEFAULT NULL,
  `ENDDATE` datetime DEFAULT NULL,
  PRIMARY KEY (`ID_JOB`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

在 Kettle 作业配置中指定该日志表路径。

2. 监控逻辑

关联 sync_timestamp 和 sync_log 表，判断是否存在未成功同步的任务。只要某张表在 sync_timestamp 中存在，但在 sync_log 中没有对应最新成功的日志，即视为异常。

检测 SQL：

SELECT COUNT(*) - (SELECT COUNT(*) FROM dbcopy.sync_timestamp) errorjob 
FROM dbcopy.sync_log a, dbcopy.sync_timestamp b 
WHERE (a.jobname, a.logdate) IN (
    SELECT jobname, MAX(logdate) 
    FROM dbcopy.sync_log 
    WHERE errors = 0 AND status = 'end' 
    GROUP BY jobname
)
AND a.jobname = b.table_name 
AND a.logdate > DATE_SUB(NOW(), INTERVAL (b.interval_seconds + 60) SECOND);

当返回值大于 0 时，说明有 JOB 执行异常，可触发告警通知运维人员介入。