Java调用Kettle工具方法

╰半夏微凉° 2023-07-18 06:20 30阅读 0赞

### Java调用Kettle工具方法 ###

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

<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][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时，每一个执行过程都会输出到所有的日志监听器中。

[https_download.csdn.net_download_xukun5137_12267874]: https://download.csdn.net/download/xukun5137/12267874