本文共 1732 字，大约阅读时间需要 5 分钟。

【Kettle】Kettle增量抽取模型

在日常的ETL过程中，常会使用用增量抽取数据，有一个简单，通用的增量抽取模型，便可以开发效率，并统一开发规范，该模型是基于时间戳的增量方式，并且有etl_logs记录抽取情况，并且具有一定的容错性。

一，抽取日志表

表字段

Id:表ID tableName:表名 min_date:最小更新时间，初始值 1980-01-01 00:00:00 max_date:最大更新时间，初始值 1980-01-01 00:00:00 status:更新状态，0-初始化，1-更新中，2-错误，3-完成 etl_date:etl开始时间 etl_date_end:etl结束时间 is_del:删除标记，0-正常，1-删除 threads:执行sql的进程数，0-没执行sql，1-正在执行etl的sql sql_text:获取最大更新时间的sql 注意： 如果表中用于增量更新的字段为update_date，则sql_text为表名。 如果表中的增量字段不为update_date，则sql_text为（select 增量字段 as update_date from tablename) t）

二，抽取过程

1，主流程

save_etl_logs：将etl_logs将日志表中的数据插入到日志历史记录表中 sql_init_etl：整个etl同步的初始化，初始化对象：status 为 0 或 3，且is_del = 0 更新内容如下sql。

update etl_logsset min_date = max_date,    etl_date = sysdate,    etl_date_end = nullwhere status in ('0','3') and is_del = 0

2，get_etl_maxdate_job

该job主要是为了获取最大的更新时间，其关键就在于etl_logs中的sql_text 主流程 转换：get_maxdate_from_logs 主要是从日志表中获取ID，SQL_TEXT，用于获取最大的更新时间。

SELECT id, sql_textFROM etl_logs WHERE status in('0','2','3')

作业：get_maxdate_job，通过sql_text获取最大更新时间，并更新到日志表中。

转换：get_variables，从结果集中获取ID，sql_text，并设置变量 转换：get_maxdate_tran，更新最大更新时间 中间的表输入sql，注意勾选替换变量

select ${id},as xid       nvl(max(update_date).to_date('1980-01-01 00:00:00','yyyy-mm-dd HH24:mi:ss')) as maxdate,       '1' as status       from ${sql_text}

3，etl_main_job

用于实现业务数据的抽取更新。 主流程 (1)，设置表id。 (2)，获取当前当前执行sql的进程，如果大于等于1，则有SQL在执行中，如果小于1，则thread+1，开始执行表插入更新。 (3)，判断插入/更新是否完成，并更新执行的状态。

转换：get_threads，获取执行SQL的thread

SQL_update_thread

update etl_logsset thread = thread+1where id = ${id}

转换：tran_xxxx_update，表的插入/更新

注意表输入的条件，增量字段的时间范围

后面更新结果的SQL，3-成功，2-失败，并且更新thread，以及接受时间。

update etl_logsset status = '3',    etl_date_end = sysdate,    thread = thread-1where id = ${id}

整个模型就是这样，由于涉及的过程比较多，刚开始搭建可能会报错，所以需要自己在根据实际情况调整。如有不对，可以拍砖指出。

转载地址：https://blog.csdn.net/yangjjuan/article/details/92213174 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！