java调用kettle执行任务或转换_Java使用TransMeta调用Kettle 执行转换文件或任务

首先下载Kettle工具，最新的版本是Kettle 5.0.1-stable,Kettle 5.0.1-stable的下载页面：Kettle 5.0.1-stable

Java调用Kettle执行任务或转换，需要使用Kettle中的jar，可以先导入lib目录中的几个基本的jar，如：kettle-core.jar、kettle-engine.jar、kettle-db.jar ，所需的jar在\lib、\libext 下面都可以找到，其它jar根据情况进行添加，本示例引用的kettle的jar如下图：

然后后编写代码测试Java调用，调用前先使用Kettle的设计器设计了一个转换，取名为mysql_to_oracle.ktr。另外，本示例使用的是Kettle 4.4.0-stable版，Kettle 4.4.0-stable的下载页： Kettle 4.4.0-stable 。

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;

import org.pentaho.di.core.KettleEnvironment;

//Java调用Kettle示例代码

// 李文锴

public class KettleTest {

public static void main(String[] args) throws KettleException {

String filename = "mysql_to_oracle.ktr";

// callNativeTrans(filename);

// executeTrans();

// executeJobs();

callNativeTransWithParam(filename);

System.out.println("ok");

}

//调用本地的转换文件

public static void callNativeTrans(String transFileName) throws KettleException {

// 初始化

EnvUtil.environmentInit();

KettleEnvironment.init();

//kettle3.2以后下面这句就不要了

//StepLoader.init();

// 转换元对象

TransMeta transMeta = new TransMeta(transFileName);

// 转换

Trans trans = new Trans(transMeta);

// 执行转换

trans.execute(null);

// 等待转换执行结束

trans.waitUntilFinished();

if ( trans.getErrors() > 0 ){

System.out.println("转换失败!");

｝

}

//调用本地的转换文件(带参数)

public static void callNativeTransWithParam(String transFileName) throws KettleException {

// 初始化

EnvUtil.environmentInit();

KettleEnvironment.init();

//kettle3.2以后下面这句就不要了

//StepLoader.init();

// 转换元对象

TransMeta transMeta = new TransMeta(transFileName);

// 转换

Trans trans = new Trans(transMeta);

String[] params = {};

// 执行转换

trans.execute(params);

// 等待转换执行结束

trans.waitUntilFinished();

}

// 执行存储在数据库资源库中的转换

public static void executeTrans() throws KettleException {

// 初始化

EnvUtil.environmentInit();

KettleEnvironment.init();

//kettle3.2以后下面这句就不要了

//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();

}

//执行本地的任务文件

public static void callNativeJob(String jobFileName) throws KettleException {

// 初始化

EnvUtil.environmentInit();

KettleEnvironment.init();

//kettle3.2以后下面这句就不要了

//StepLoader.init();

JobEntryLoader.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();

}

//执行数据库资源库中的任务

public static void executeJobs() throws KettleException {

// 初始化

EnvUtil.environmentInit();

JobEntryLoader.init();

KettleEnvironment.init();

//kettle3.2以后下面这句就不要了

//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();

}

}

参考资料地址：http://blog.csdn.net/scorpio3k/article/details/7869196  作者：scorpio3k