Java调用Kettle工具方法

这里需要的主要依赖包下面已列出，其他的运行时缺什么补什么就行：

pentaho-kettle

kettle-core

7.1.0.0-12





pentaho-kettle

kettle-dbdialog

7.1.0.0-12




pentaho-kettle

kettle-engine
7.1.0.0-12
/dependency>
br />
dependency>

pentaho-kettle
metastore
/artifactId>
br />
version>7.1.0.0-12
/version>
br />
br />
br />
KettleLoggingEventListener listener = null;
br />
try {
br />
br />
log.debug("开始执行Kettler任务");
br />
log.debug("创建TransMeta源数据对象");
br />
TransMeta transMeta = new TransMeta(KETTLE_KET_DIR + "\\" + ktrName);
br />
log.debug("传入KTR文件所需要的参数值");
br />
if (null != params) {
br />
br />
for (Map.Entry
String, String>entry : params.entrySet()) {


transMeta.setVariable(entry.getKey(), entry.getValue());

}

}

Trans transformation = new Trans(transMeta);

log.debug("设置Kettle日志级别");

transformation.setLogLevel(LogLevel.BASIC);

log.debug("创建日志监听程序");

listener = new ConsoleLoggingEventListener();

KettleLogStore.getAppender().addLoggingEventListener(listener);

log.debug("开始执行Kettle文件");

transformation.prepareExecution(null);

transformation.startThreads();

log.debug("等待Kettle执行完成");

transformation.waitUntilFinished();

if (transformation.getErrors() >0) {


log.error("Kettle执行过程中发生异常");

throw new RuntimeException("Kettle执行过程中发生异常,请查看日志信息");

} else {


log.debug("Kettle执行成功");

}

} catch (Exception e) {


log.error("Kettle执行失败", e);

} finally {


if(null != listener){


KettleLogStore.getAppender().removeLoggingEventListener(listener);

}

log.debug("Kettle执行结束");

}

}

/** * 调用Kettle执行作业文件 * * @methodname runJob * @param kjbName 作业文件名称 * @param params 作业参数 * * @return void * @date 2020/3/24 18:33 */

public static void runJob(String kjbName, Map
 params) throws Exception { 



KettleLoggingEventListener kettlelog = null;

try { 



log.debug("开始执行Kettle任务");

log.debug("创建Job的源数据对象");

JobMeta jobMeta = new JobMeta(KETTLE_KET_DIR + "\\" + kjbName, null);

log.debug("创建Job对象");

Job job = new Job(null, jobMeta);

log.debug("传入作业执行需要的参数");

if (null != params) { 



for (Map.Entry entry : params.entrySet()) { 



job.setVariable(entry.getKey(), entry.getValue());

}

}

log.debug("设置Kettle日志级别");

job.setLogLevel(LogLevel.BASIC);

log.debug("创建日志监听程序");

kettlelog = new ConsoleLoggingEventListener();

KettleLogStore.getAppender().addLoggingEventListener(kettlelog);

log.debug("开始执行Job");

job.start();

log.debug("等待Job执行完成");

job.waitUntilFinished();

job.setFinished(true);

if (job.getErrors() > 0) { 



log.error("Job执行过程中发生异常");

throw new RuntimeException("Job执行过程中发生异常,请查看日志信息");

} else { 



log.debug("Job执行成功");

}

} catch (Exception e) { 



log.error("Kettle执行失败", e);

} finally { 



if(null != kettlelog ){ 



KettleLogStore.getAppender().removeLoggingEventListener(listener);

}

log.debug("Kettle执行结束");

}

}

}

对于日志监听器需要注意的地方，日志监听器不针对某一个执行过程，如果并发执行Kettle时，每一个执行过程都会输出到所有的日志监听器中。