**
kettle数据转换工具使用案例Kettle简介
Kettle是一款开源的etl工具，纯java编写，绿色无需安装，数据抽取高效稳定。
Kettle中有两种脚本文件，transformation和job，transformation完成针对数据的基础转换，job则完成整个工作流的控制。
kettle 部署运行
将pdi-7.1.0.0-12.zip文件夹拷贝到本地路径，例如D 盘根目录。
双击运行kettle文件夹下的spoon.bat文件，出现kettle欢迎界面，如果无法打开请检查本地是否已经安装jdk运行环境：
进入主界面:
新建转换和作业
点击页面左上角的创建一个新的转换，点击保存到本地路径，例如保存到D:/etltest下，保存文件名为EtltestTrans，kettle默认转换文件保存后后缀名为ktr。
点击页面左上角的创建一个新的作业，点击保存到本地路径，例如保存到D:/etltest下，保存文件名为EtltestJob，kettle默认job文件保存后后缀名为kjb。
创建数据库连接
在转换页面下，点击左边的主对象树，双击【DB连接】，进行数据库连接配置。
推荐使用Generic Database，因为mysql6以上版本如果使用mysql-connector-java-8.0.19.jar这个驱动，会提示找不到驱动包，用低版本的驱动转换的时候会报错。
点击测试，出现下面的弹框，提示没有对应的数据库驱动：
将提示的驱动包放入kettle文件夹下面的lib目录，重启kettle。点击测试，弹出如下提示表示连接成功。
案例一：两表数据同步
目标：将不同数据库里面的两张表进行数据同步，每分钟同步一次
点击文件-新建-转换，在左侧核心对象tab页下面将输入-表输入，输出-插入/更新，这两个对象拖到右侧编辑栏中，建立连接，如图：
【表输入】配置如下：
【插入/更新】配置如下：
Kettle执行这个步骤是需要两个数据流对比，其中一个是目标数据库，你在目标表 里面指定的，它放在用来查询的关键字左边的表字段里面的，另外一个数据流就是你在前一个步骤传进来的，它放在用来查询的关键字 的右边，Kettle首先用你传进来的key 在数据库中查询这些记录，如果没有找到，它就插入一条记录，所有的值都跟你原来的值相同，如果根据这个key找到了这条记录，kettle会比较这两条记录，根据你指定update field 来比较，如果数据完全一样，kettle就什么都不做，如果记录不完全一样，kettle就执行一个update 步骤。
备注：主键被修改得数据认为是新记录,删除的数据由在仓库中需要保留无需考虑
配置完成后点击运行这个转换按钮，查看日志如下，表示转换成功。
点击文件-新建-作业，在左侧核心对象tab页下面将通用-start，通用-转换，这两个对象拖到右侧编辑栏中，建立连接
【START】配置如下：
【转换】配置如下：
点击RUN，这样就完成了一个每分钟执行一次的数据同步任务。
案例二：Java代码组件应用目标：现有一张order订单表，将其根据订单号显示的年份分表存储到按年划分的三张订单表中
源数据库订单表DDL：
CREATE TABLE order (
id bigint(20) NOT NULL AUTO_INCREMENT,
order_num varchar(20) DEFAULT NULL,
content varchar(100) DEFAULT NULL,
create_time datetime DEFAULT NULL,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8;
目标数据库新增三张订单表DDL：
CREATE TABLE order_2020 (
id bigint(20) NOT NULL AUTO_INCREMENT,
order_num varchar(20) DEFAULT NULL,
content varchar(100) DEFAULT NULL,
create_time datetime DEFAULT NULL,
PRIMARY KEY (id)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8;
其他表以此类推
点击文件-新建-转换，在左侧核心对象tab页下面将输入-表输入，脚本-java代码，输出-表输出，这三个对象拖到右侧编辑栏中，建立连接，如图：
【表输入】配置如下：
【java代码】配置如下：
public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws Exception { Object[] r = getRow(); //这里获取到上一个节点流转的行数据 if (r == null) { setOutputDone(); return false; } r = createOutputRow(r, data.outputRowMeta.size()); //创建输出行数据 String orderNum = get(Fields.In, "ORDER_NUM").getString(r); String content = get(Fields.In, "CONTENT").getString(r); String createTime = get(Fields.In, "CREATE_TIME").getString(r); String year = orderNum.substring(0,4); get(Fields.Out,"table_name").setValue(r,"order_"+year); //将订单根据订单号前四位年份生成新的表名，放入输出行数据中 putRow(data.outputRowMeta, r); return true;}
【表输出】配置如下：
配置完成后点击运行这个转换按钮，查看日志无报错，然后去新表里面查看数据，看到订单数据已经被拆分到了三张不同表中，则转换成功。
