Kettle与Java集成——Java代码调取运行资源库的Transformation

下面代码是Java代码调用Kettle4.0版本的Transformation

1、Java读取资源库运行Transformation

(1)目录类型资源库Java代码

 

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.repository.filerep.KettleFileRepository;
import org.pentaho.di.repository.filerep.KettleFileRepositoryMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
public class ReaderTransFromRep {
    private static String tranName = "IM_KSDM";    // 传输名称
    public static void main(String[] args) throws KettleException {
        Trans trans=null;
        try {
            // 初始化
            KettleEnvironment.init();这个是4.0版本的初始化，和3.x版本的不同
            // 资源库元对象
            KettleFileRepositoryMeta repinfo = new KettleFileRepositoryMeta("","","数据采集","file:///E:/Workspaces/Kettle");
            // 文件形式的资源库
            KettleFileRepository rep = new KettleFileRepository();
            rep.init(repinfo);
            // 转换元对象
            if(tranName!=null){
                TransMeta transMetadel = rep.loadTransformation(rep.getTransformationID(tranName, null), null);
                // 转换
                trans = new Trans(transMetadel);
                // 执行转换
                trans.execute(null);
                // 等待转换执行结束
                trans.waitUntilFinished();
                //抛出异常
                if(trans.getErrors()>0){
                    throw new Exception("There are errors during transformation exception!(传输过程中发生异常)");
                }
            }else{
                throw new KettleException("传输名为空！");
            }
        } catch (Exception e) {
            if(trans!=null){
                trans.stopAll();
            }
            e.printStackTrace();
            throw new KettleException(e);
        }
    }
                      }

 

 

(2)数据库类型资源库

 

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
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;

public class ReaderTransFromDBRep {
    private static String transName="t1";
    public static void main(String[] args) {

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