kettle 更新/插入组件 JAVA API

package com.example.fg.kettle;


import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.ObjectLocationSpecificationMethod;
import org.pentaho.di.core.RowMetaAndData;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobHopMeta;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.job.entries.special.JobEntrySpecial;
import org.pentaho.di.job.entries.success.JobEntrySuccess;
import org.pentaho.di.job.entries.trans.JobEntryTrans;
import org.pentaho.di.job.entry.JobEntryCopy;
import org.pentaho.di.repository.ObjectId;
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.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;

import javax.validation.constraints.Null;
import java.util.List;

/**
 * @author cgm
 * @date 2019年7月8日
 */
public class KettleJobCronTest {
    public static void main(String[] args) throws Exception {
        runSyncJob();

    }

    public static void runSyncJob() throws KettleException {
        String transName = "cgmTransName";
        String repositoryName = "Repository";
        String jobName = "cgmTransName";
        KettleDatabaseRepository kettleDatabaseRepository  = RepositoryCon(repositoryName);
        RepositoryDirectoryInterface dir = kettleDatabaseRepository.loadRepositoryDirectoryTree().findDirectory("/");
        ObjectId objectId = kettleDatabaseRepository.getTransformationID(transName, dir);
        TransMeta transMeta;
        if (objectId == null) {
            transMeta = new TransMeta();
            // 设置转化的名称
            transMeta.setName(transName);
            // 添加转换的数据库连接
            transMeta.addDatabase(new DatabaseMeta("fromDbName", "mysql", "Native(JDBC)", "127.0.0.1",
                    "test?useSSL=false", "3306", "root", "toor@1234"));
            transMeta.addDatabase(new DatabaseMeta("toDbName", "mysql", "Native(JDBC)", "127.0.0.1", "test?useSSL=false",
                    "3306", "root", "toor@1234"));

            // 新建一个表输入步骤(TableInputMeta)
            TableInputMeta tableInputMeta = new TableInputMeta();
            // 设置步骤1的数据库连接
            tableInputMeta.setDatabaseMeta(transMeta.findDatabase("fromDbName"));
            // 设置步骤1中的sql
            tableInputMeta.setSQL("SELECT id1 as id2 ,name1 as name2 FROM from_user");
            // 设置步骤名称
            StepMeta step1 = new StepMeta("step1name", tableInputMeta);
            transMeta.addStep(step1);

            // 新建一个插入/更新步骤
            InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
            // 设置步骤2的数据库连接
            insertUpdateMeta.setDatabaseMeta(transMeta.findDatabase("toDbName"));
            // 设置目标表
            insertUpdateMeta.setTableName("to_user");
            // 设置用来查询的关键字
            insertUpdateMeta.setKeyLookup(new String[]{"id2"});
            insertUpdateMeta.setKeyCondition(new String[]{"="});
            insertUpdateMeta.setKeyStream(new String[]{"id2"});
            insertUpdateMeta.setKeyStream2(new String[]{""});// 一定要加上
            // 设置要更新的字段
            String[] updatelookup = {"id2", "name2"};
            String[] updateStream = {"id2", "name2"};
            Boolean[] updateOrNot = {false, true};
            // 设置表字段
            insertUpdateMeta.setUpdateLookup(updatelookup);
            // 设置流字段
            insertUpdateMeta.setUpdateStream(updateStream);
            // 设置是否更新
            insertUpdateMeta.setUpdate(updateOrNot);
            // 设置步骤2的名称
            StepMeta step2 = new StepMeta("step2name", insertUpdateMeta);
            transMeta.addStep(step2);

            TransHopMeta transHopMeta = new TransHopMeta(step1, step2);
            transMeta.addTransHop(transHopMeta);


            transMeta.setRepositoryDirectory(dir);
            kettleDatabaseRepository.save(transMeta, null);
        }

        ObjectId id = kettleDatabaseRepository.getJobId(jobName, dir);
        if (id == null) {
            JobMeta jobMeta = new JobMeta();
            jobMeta.setName(jobName);
            jobMeta.setJobstatus(0);

            JobEntrySpecial jobEntrySpecial = new JobEntrySpecial();
            jobEntrySpecial.setName("START");
            jobEntrySpecial.setStart(true);
            jobEntrySpecial.setRepeat(true);
            jobEntrySpecial.setIntervalSeconds(3);
            JobEntryCopy start = new JobEntryCopy(jobEntrySpecial);
            start.setDrawn();
            start.setLocation(100, 200);

            JobEntryTrans jobEntryTrans = new JobEntryTrans();
            ObjectLocationSpecificationMethod SpecMethod = ObjectLocationSpecificationMethod.getSpecificationMethodByCode(repositoryName);
            jobEntryTrans.setSpecificationMethod(SpecMethod);
            jobEntryTrans.setRepository(kettleDatabaseRepository);
            jobEntryTrans.setDirectory("/");
            jobEntryTrans.setTransname(transName);
            JobEntryCopy trans1 = new JobEntryCopy(jobEntryTrans);
            trans1.setName("JTrans");
            trans1.setDrawn(true);
            trans1.setLocation(200, 300);

            JobEntrySuccess jobEntrySuccess = new JobEntrySuccess();
            jobEntrySuccess.setName("Success");
            JobEntryCopy success = new JobEntryCopy(jobEntrySuccess);
            success.setDrawn();
            success.setLocation(300, 400);

            jobMeta.addJobEntry(start);
            jobMeta.addJobEntry(trans1);
            jobMeta.addJobEntry(success);
            
            jobMeta.addJobHop(new JobHopMeta(start, trans1));
            jobMeta.addJobHop(new JobHopMeta(trans1, success));

            jobMeta.setRepositoryDirectory(dir);
            kettleDatabaseRepository.save(jobMeta, null);
        }

        JobMeta jobMetaObj = kettleDatabaseRepository.loadJob(id, null);
        Job job = new Job(kettleDatabaseRepository, jobMetaObj);
        job.run();
        job.waitUntilFinished();
        String strLog = KettleLogStore.getAppender().getBuffer().toString();
        System.out.println("==========开始打印日志==========");

        System.out.println(KettleLogStore.getAppender().getBuffer().toString());
        System.out.println("==========日志打印结束==========");
        System.out.println("getLastProcessed:" + job.getStatus());

        String substring = strLog.substring(strLog.lastIndexOf("I=") + 1);
        String successCount = substring.substring(substring.lastIndexOf("W=") + 2, substring.lastIndexOf("W=") + 3);
        System.out.println("成功数：" + successCount);

        System.out.println("errors:" + job.getErrors());
        if (job.getErrors() != 0) {
            System.out.println("执行失败！");
        }


    }

    public static KettleDatabaseRepository RepositoryCon(String repositoryName) throws KettleException {
        // 初始化环境
        if (!KettleEnvironment.isInitialized()) {
            try {
                KettleEnvironment.init();
            } catch (KettleException e) {
                e.printStackTrace();
            }
        }
        // 数据库连接元对象
        // （kettle数据库连接名称(KETTLE工具右上角显示)，资源库类型，连接方式，IP，数据库名，端口，用户名，密码） //cgmRepositoryConn
        DatabaseMeta databaseMeta = new DatabaseMeta(repositoryName, "mysql", "Native(JDBC)", "127.0.0.1",
                "test", "3306", "root", "toor@1234");
        KettleDatabaseRepositoryMeta kettleDatabaseRepositoryMeta = new KettleDatabaseRepositoryMeta();
        kettleDatabaseRepositoryMeta.setConnection(databaseMeta);
        KettleDatabaseRepository kettleDatabaseRepository = new KettleDatabaseRepository();
        kettleDatabaseRepository.init(kettleDatabaseRepositoryMeta);
        kettleDatabaseRepository.connect("admin", "admin");
        if (kettleDatabaseRepository.isConnected()) {
            System.out.println("连接成功");
            return kettleDatabaseRepository;
        } else {
            System.out.println("连接失败");
            return null;
        }
    }



}