使用kettle做数据ETL

首先下载kettle，我下载的是7.1.注意使用7.1的版本需要JDK1.8以上，另外Windows下需要IE11，如果没有可能会出现一些嵌入式WEB页面加载不出来的情况，导致有些功能无法使用。

在Windows下运行spoon.bat,打开spoon。spoon是kettle的图形开发工具，可以在这上面开发转换和job。

spoon界面

如果要是使用资源库（资源库是自定义的一些资源可以和别人共享，或者可以一次创建资源以便供后面多次使用，创建的转换和job），要使用资源库，点击右上角Connect，然后在弹出窗中点击getstart(如果在win下弹出窗无内容，确认是否装了IE11)，新建一个数据库资源库（也可以新建文件资源库）。新建了数据库资源库后，再连接到资源库，默认的用户名和密码都是admin。

人后新建的转换和job都会保存在资源库里面了。

在java中调用：
我的需求是要在Java程序中控制kettle任务的执行,在spoon中开发测试好之后，在Java中从资源库中获取转换和任务执行。
相关代码如下：

package com.wdkj.test.kettle;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.Repository;
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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KettleTest {

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

    private static KettleDatabaseRepository repository;
    private static RepositoryDirectoryInterface directory;

    public static void main(String... args){
        initRepo();
        try {
            runTrans();
        }catch (Exception e){
            log.error("run trans error", e);
        }
    }

    public static void initRepo(){
        try {
            KettleEnvironment.init();
            repository = new KettleDatabaseRepository();
            DatabaseMeta databaseMeta = new DatabaseMeta("kettle_repo", "MySql", "Native", "192.168.0.101",
                    "kettle", "3306",
                    "manager", "manager");//资源库数据库地址，我这里采用oracle数据库
            KettleDatabaseRepositoryMeta kettleDatabaseMeta = new KettleDatabaseRepositoryMeta("kettle_repo", "kettle_repo",
                    "Transformation description", databaseMeta);
            repository.init(kettleDatabaseMeta);
            repository.connect("admin", "admin");//资源库用户名和密码
            directory = repository.loadRepositoryDirectoryTree();

        } catch (KettleException e) {
            log.error("init kettle error :", e);
        }
    }

    public static void runTrans() throws KettleException {
        //根据变量查找到模型所在的目录对象,此步骤很重要。
        RepositoryDirectoryInterface directory = repository.findDirectory("/");
        //创建ktr元对象
        TransMeta transformationMeta = repository.loadTransformation("marriage_trans", directory, null, true, null);
        
        transformationMeta.setParameterValue("original_database_password", "gerakan");

        //创建ktr
        Trans trans = new Trans(transformationMeta);

        //执行ktr
        String[] params = {"original_database_port"};
        trans.execute( params);
        //等待执行完毕
        trans.waitUntilFinished();
        if (trans.getErrors() > 0) {
            log.info("trans executed failed");
        } else {
            log.info("trans executed OK");
        }
    }

    public static void runJob() throws KettleException {
        JobMeta jobMeta = ((Repository) repository).loadJob("job", directory, null, null);
        Job job = new Job(repository, jobMeta);
        job.setVariable("id", 1);
        job.start();
        job.waitUntilFinished();
        if (job.getErrors() > 0) {
            log.info("trans executed failed");
        } else {
            log.info("trans executed OK");
        }
    }
}

另外在Java调用数据库资源库的时候找不到IMetaStore这个类的错误,把metastore的jar导入就行。

在spoon开发和测试环境和生产环境中要使用不同的配置，只需要设置参数值即可。

如果需要执行时设置变量,即使用设置变量控件，需要新建job，在job中最前的转换中设置变量，后面的转换才能使用设置的变量.