kettle ETL java 调用 kettle job 或 transfer

亦凉 2022-01-26 06:17 403阅读 0赞

代码如下：调用kettle 4.2版本

package test;  
  
  
import org.apache.commons.lang.StringUtils;  
import org.apache.commons.vfs.FileSelector;  
import org.pentaho.di.core.KettleEnvironment;  
import org.pentaho.di.core.database.DatabaseMeta;  
import org.pentaho.di.core.exception.KettleException;  
import org.pentaho.di.core.util.EnvUtil;  
import org.pentaho.di.job.Job;  
import org.pentaho.di.job.JobMeta;  
import org.pentaho.di.repository.RepositoryDirectory;  
import org.pentaho.di.repository.RepositoryDirectoryInterface;  
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;  
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;  
import org.pentaho.di.trans.Trans;  
import org.pentaho.di.trans.TransMeta;  
  
  
import com.longtech.configuration.BaseConfiguration;  
import com.platform.common.util.LoggerUtil;  
  
  
/\*\*  
 \*   
 \* @Description:  java调用kettle 数据库型资料库中的转换  
 \* @author chen 此类是调用kettle 4.2版本的测试类  
 \* @version 1.0,   
 \* @date 2013-5-23 下午02:40:50  
 \*/  
public class ExecRepositoryTrans \{  
/\*\*  
\* 本测试类慎用！！！！！！！  
\* @param args  
\*/  
public static void main(String\[\] args) \{  
//D:\\java\\workspace\\sics\\src\\config\\kettle  
String classesdir = BaseConfiguration.getString("classesdir");  
String jobName = BaseConfiguration.getString("kettle.path.file");  
if(StringUtils.isNotEmpty(jobName))\{  
LoggerUtil.info(ExecRepositoryTrans.class, "执行kettle job start-------------------");  
  
runJob(classesdir + jobName);  
  
LoggerUtil.info(ExecRepositoryTrans.class, "执行kettle job end -------------------");  
\}else\{  
  
LoggerUtil.info(ExecRepositoryTrans.class, "请配置kettle.path.file job 的路径文件");  
\}  
  
\}  
/\*\*  
\*java 调用kettle 转换   
\*/  
public static void runTransfer()\{  
Trans trans=null;    
FileSelector f = null;  
try \{  
// 初始化    
            String fName= "D:\\\\kettle\\\\informix\_to\_am\_4.ktr";  
            // 转换元对象    
        KettleEnvironment.init();//初始化  
        EnvUtil.environmentInit();  
            TransMeta transMeta = new TransMeta(fName);  
            // 转换    
            trans = new Trans(transMeta);    
            // 执行转换    
            trans.execute(null);     
            // 等待转换执行结束    
            trans.waitUntilFinished();    
            //抛出异常    
            if(trans.getErrors()>0)\{    
                throw new Exception("There are errors during transformation exception!(传输过程中发生异常)");    
            \}    
\} catch (Exception e) \{  
e.printStackTrace();  
\}  
\}  
/\*\*  
\* java 调用 kettle 的job  
\* @param jobname 如： String fName= "D:\\\\kettle\\\\informix\_to\_am\_4.ktr";  
\*/  
public static void runJob(String jobname)\{    
try \{     
KettleEnvironment.init();     
//jobname 是Job脚本的路径及名称     
JobMeta jobMeta = new JobMeta(jobname, null);     
Job job = new Job(null, jobMeta);     
//向Job 脚本传递参数，脚本中获取参数值：$\{参数名\}     
//job.setVariable(paraname, paravalue);     
 job.start();     
 job.waitUntilFinished();     
 if (job.getErrors() > 0) \{     
 throw new Exception("There are errors during job exception!(执行job发生异常)");    
 \}     
\} catch (Exception e) \{     
e.printStackTrace();  
\}     
\}  
/\*\*  
\* 配置数据源 调用资源库中的相关job 、transfer  
\*/  
public static void dbResource()\{  
String transName="t1";    
           
       try \{    
           KettleEnvironment.init();    
           DatabaseMeta dataMeta = new DatabaseMeta("KettleDBRep","MSSQL","Native","127.0.0.1","etl","1433","sa","bsoft");    
           KettleDatabaseRepositoryMeta repInfo = new KettleDatabaseRepositoryMeta();    
           repInfo.setConnection(dataMeta);    
           KettleDatabaseRepository rep = new KettleDatabaseRepository();    
           rep.init(repInfo);    
           rep.connect("admin", "admin");    
               
           RepositoryDirectoryInterface dir = new RepositoryDirectory();    
           dir.setObjectId(rep.getRootDirectoryID());    
               
           TransMeta tranMeta = rep.loadTransformation(rep.getTransformationID(transName, dir), null);    
           Trans trans = new Trans(tranMeta);    
           trans.execute(null);    
           trans.waitUntilFinished();    
       \} catch (KettleException e) \{    
           e.printStackTrace();    
       \}    
\}  
  
  
\}