kettle工具中嵌入java代码_kettle设置变量

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

<dependency>
    <groupId>pentaho-kettle</groupId>
    <artifactId>kettle-core</artifactId>
    <version>7.1.0.0-12</version>
</dependency>
<dependency>
    <groupId>pentaho-kettle</groupId>
    <artifactId>kettle-dbdialog</artifactId>
    <version>7.1.0.0-12</version>
</dependency>
<dependency>
    <groupId>pentaho-kettle</groupId>
    <artifactId>kettle-engine</artifactId>
    <version>7.1.0.0-12</version>
</dependency>
<dependency>
    <groupId>pentaho-kettle</groupId>
    <artifactId>metastore</artifactId>
    <version>7.1.0.0-12</version>
</dependency>

插件地址 https://download.csdn.net/download/xukun5137/12267874

/** * Java调用Kettle工具类 * * @classname KettleUtil * @date 2020/3/24 16:24 */
public final class KettleUtil { 
   

    private static final Logger log = LoggerFactory.getLogger(KettleUtil.class);

    private static final String KETTLE_BASE_DIR = "D:\\kettle";
    private static final String KETTLE_KET_DIR = KETTLE_BASE_DIR + "\\ktr";
    private static final String KETTLE_PLUGIN_DIR = KETTLE_BASE_DIR + "\\plugins";

    static { 
   
        try { 
   
            log.debug("加载Kettle插件");
            StepPluginType.getInstance().getPluginFolders().add(new PluginFolder(KETTLE_PLUGIN_DIR, false, true));
            log.debug("Kettle插件加载成功");
            log.debug("初始化Kettle插件环境");
            KettleEnvironment.init();
            log.debug("Kettle插件环境初始化完成");
        } catch (KettleException e) { 
   
            log.error("Kettle插件加载初始化失败");
        }
    }

    /** * 调用Kettle插件执行KTR文件 * * @param ktrName 目标KTR文件名称 * @param params KTR文件所需要的参数信息 * * @return void * @date 2020/3/24 16:31 */
    public static void runKtr(String ktrName, Map<String, String> params) { 
   
    	KettleLoggingEventListener listener = null;
        try { 
   
            log.debug("开始执行Kettler任务");

            log.debug("创建TransMeta源数据对象");
            TransMeta transMeta = new TransMeta(KETTLE_KET_DIR + "\\" + ktrName);

            log.debug("传入KTR文件所需要的参数值");
            if (null != params) { 
   
                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<String, String> 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<String, String> 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时，每一个执行过程都会输出到所有的日志监听器中。