Kettle框架搭建以及java代码调用作业和转换
Kettle框架搭建
下载安装
可以从http://kettle.pentaho.org下载最新版的 Kettle软件 ,同时,Kettle 是绿色软件,下载后,解压到任意目录即可。
目前Kettle的最新版本是7.1。
由于Kettle 是采用java 编写,因此需要在本地有JVM 的运行环境。
安装完成之后,点击目录下面的kettle.exe 或者spoon.bat 即可启动kettle 。在启动kettle 的时候,会弹出对话框,让用户选择建立一个资源库。
资源库:是用来保存转换任务的, 它用以记录我们的操作步骤和相关的日志,转换,JOB 等信息。用户通过图形界面创建的的转换任务可以保存在资源库中。资源库可以是各种常见的数据库,用户通过用户名/ 密码来访问资源库中的资源,默认的用户名/ 密码是admin/admin. 资源库并不是必须的,如果没有资源库,用户还可以把转换任务保存在 xml 文件中。
配置环境变量
前提是配置好Java的环境变量,因为他是java编写,需要本地的JVM的运行环境
配置Java环境变量可参考:
java实战(一)———–jdk环境配置
http://blog.****.net/zzq900503/article/details/9770237
在系统的环境变量中添加KETTLE_HOME变量,目录指向kettle的安装目录:D:kettledata-integration(具体以安装路径为准)
新建系统变量:KETTLE_HOME
变量值: D:kettledata-integration(具体以安装路径为准,Kettle的解压路径,直到Kettle.exe所在目录)
选择PATH添加环境变量:
变量名:PATH
变量值:% KETTLE_HOME%;
spoon操作界面
打开spoon.bat文件所在位置如下图
spoon基本操作界面
在spoon的操作界面中可以创建转换和工作
Kettle的基本概念
作业(job)
负责将[转换]组织在一起进而完成某一块工作,通常我们需要把一个大的任务分解成几个逻辑上隔离的作业,当这几个作业都完成了,也就说明这项任务完成了。
1.Job Entry:一个Job Entry 是一个任务的一部分,它执行某些内容。
2.Hop:一个Hop 代表两个步骤之间的一个或者多个数据流。一个Hop 总是代表着两个Job Entry 之间的连接,并且能够被原始的Job Entry 设置,无条件的执行下一个Job Entry,
直到执行成功或者失败。
3.Note:一个Note 是一个任务附加的文本注释信息。
转换(Transformation)
定义对数据操作的容器,数据操作就是数据从输入到输出的一个过程,可以理解为比作业粒度更小一级的容器,我们将任务分解成作业,然后需要将作业分解成一个或多个转换,每个转换只完成一部分工作。
1.Value:Value 是行的一部分,并且是包含以下类型的的数据:Strings、floating point Numbers、unlimited precision BigNumbers、Integers、Dates、或者Boolean。
2.Row:一行包含0 个或者多个Values。
3.Output Stream:一个Output Stream 是离开一个步骤时的行的堆栈。
4.Input Stream:一个Input Stream 是进入一个步骤时的行的堆栈。
5.Step:转换的一个步骤,可以是一个Stream或是其他元素。
6.Hop:一个Hop 代表两个步骤之间的一个或者多个数据流。一个Hop 总是代表着一个步骤的输出流和一个步骤的输入流。
7.Note:一个Note 是一个转换附加的文本注释信息。
创建简单的转换,首先配置数据库连接,一个转换中可以创建多个数据库连接,而且是不同类型的数据库也都可以,方便从一个数据库中抽取数据到另一个数据库中
左上角“核心对象”选项卡。展开输入选择->表输入,将表输入拖入到右侧控制台,表输出也是做相同的操作。
将鼠标放在表输入上停几秒会出现如图所示的选择框,建立表输入与表输出的连接关系。
双击表输入,然后给步骤自定义名称,并选择数据库连接,然后点击获取SQL查询语句按钮,选在数据库中需要操作的表,下方SQL展示框中会显示简单的sql语句,可以手动修改相应的sql语句。
Excel表输出:选择要输出的文件名。扩展名可以是xls也可以是xlsx
然后点击字段:选择数据库表中要导出到excel中的字段名称,类型和格式。可以导出指定字段也可以全部导出
以上转换操作做完以后点击执行按钮即可执行当前转换
作业的概念是可以将多个转换,或者多个作业串在一起共同执行。
在实际开发中通常会需要将转换或者作业嵌入到代码中,也就是java代码调用我们编辑好的作业和转换,下面写一个实际的调用实例
(1)首先需要导入kettle的相关jar包,如下图有些是数据库的jar包,作业或转换中所连接的数据库jar包都要有,所需jar包下载地址https://download.****.net/download/m0_37027631/10340506
(2)KettleExecu.java类,解释写在了代码中,下面有测试方法
package com.kettle;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogWriter;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobEntryLoader;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.repository.RepositoryDirectory;
import org.pentaho.di.repository.RepositoryMeta;
import org.pentaho.di.repository.UserInfo;
import org.pentaho.di.trans.StepLoader;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
/**
*
* @author lyd
* @date 2018年3月13日
*/
public class KettleExecu {
public static void main(String[] args) throws KettleException {
String filename = "D:/projects/oracleToexcel.ktr";
// callNativeTrans(filename);
// executeTrans();
// executeJobs();
callNativeTransWithParam(filename);
System.out.println("mysql、oracle数据库指定表导出到excel成功!");
}
/**
* 调用本地的转换文件
*
* @Description:
* @param transFileName
* @throws KettleException
* @author lyd
* @since:2018 下午02:58:41
*/
public static void callNativeTrans(String transFileName) throws KettleException {
// 初始化
EnvUtil.environmentInit();
StepLoader.init();
// 转换元对象
TransMeta transMeta = new TransMeta(transFileName);
// 转换
Trans trans = new Trans(transMeta);
// 执行转换
trans.execute(null);
// 等待转换执行结束
trans.waitUntilFinished();
}
/**
* 调用本地的转换文件(带参数)
* @Description:
* @param transFileName
* @throws KettleException
* @author lyd
* @since:2018年3月13日
*/
public static void callNativeTransWithParam(String transFileName) throws KettleException {
// 初始化
EnvUtil.environmentInit();
StepLoader.init();
// 转换元对象
TransMeta transMeta = new TransMeta(transFileName);
// 转换
Trans trans = new Trans(transMeta);
String[] params = {};
// 执行转换
trans.execute(params);
// 等待转换执行结束
trans.waitUntilFinished();
}
/**
* 执行存储在数据库资源库中的转换
*
* @Description:
* @throws KettleException
* @author lyd
* @since:2018年3月13日
*/
public static void executeTrans() throws KettleException {
// 初始化
EnvUtil.environmentInit();
StepLoader.init();
// 日志
LogWriter log = LogWriter.getInstance("TransTest.log", true, LogWriter.LOG_LEVEL_DEBUG);
// 用户
UserInfo userInfo = new UserInfo();
userInfo.setLogin("admin");
userInfo.setPassword("admin");
// 数据库连接元对象(连接名称,不必与kettle中配置的保持一致:数据库类型:连接方式(kettle支持的连接方式):资源库IP:资源库实例名:资源库端口:资源库用户名:资源库用户密码)
DatabaseMeta connection = new DatabaseMeta("", "Oracle", "Native", "192.168.3.232", "NSDEV", "1521", "nstcsa3441", "671468");
// 资源库元对象
RepositoryMeta repinfo = new RepositoryMeta();
repinfo.setConnection(connection);
// 资源库
Repository rep = new Repository(log, repinfo, userInfo);
// 连接资源库
rep.connect("");
// 资源库目录对象
RepositoryDirectory dir = new RepositoryDirectory(rep);
// 转换元对象
TransMeta transMeta = new TransMeta(rep, "凭证(N9->EVC2)", dir);
// 转换
Trans trans = new Trans(transMeta);
// 执行转换
trans.execute(null);
// 等待转换执行结束
trans.waitUntilFinished();
}
/**
* 执行本地的任务文件
*
* @Description:
* @param jobFileName
* @throws KettleException
* @author lyd
* @since:2018年3月13日
*/
public static void callNativeJob(String jobFileName) throws KettleException {
// 初始化
EnvUtil.environmentInit();
JobEntryLoader.init();
StepLoader.init();
// 日志
LogWriter log = LogWriter.getInstance("TransTest.log", true, LogWriter.LOG_LEVEL_DETAILED);
// job元对象
JobMeta jobMeta = new JobMeta(log, jobFileName, null);
// job
Job job = new Job(log, StepLoader.getInstance(), null, jobMeta);
jobMeta.setInternalKettleVariables(job);
// 执行job
job.execute();
// 等待job执行结束
job.waitUntilFinished();
}
/**
* 执行数据库资源库中的任务
*
* @Description:
* @throws KettleException
* @author lyd
* @since:2018年3月13日
*/
public static void executeJobs() throws KettleException {
// 初始化
EnvUtil.environmentInit();
JobEntryLoader.init();
StepLoader.init();
// 日志
LogWriter log = LogWriter.getInstance("TransTest.log", true, LogWriter.LOG_LEVEL_DETAILED);
// 用户
UserInfo userInfo = new UserInfo();
userInfo.setLogin("admin");
userInfo.setPassword("admin");
// 数据库连接元对象
DatabaseMeta connection = new DatabaseMeta("", "Oracle", "Native", "192.168.3.232", "NSDEV", "1521", "nstcsa3441", "671468");
// 资源库元对象
RepositoryMeta repinfo = new RepositoryMeta();
repinfo.setConnection(connection);
// 资源库
Repository rep = new Repository(log, repinfo, userInfo);
// 连接资源库
rep.connect("");
// 资源库目录对象
RepositoryDirectory dir = new RepositoryDirectory(rep);
// 步骤加载对象
StepLoader steploader = StepLoader.getInstance();
// job元对象
JobMeta jobmeta = new JobMeta(log, rep, "4.账户每日余额", dir);
// job
Job job = new Job(log, steploader, rep, jobmeta);
// 执行job
job.execute();
// 等待job执行结束
job.waitUntilFinished();
}
}