java调用kettle文件,根据环境修改数据源

在使用kettle可视化工具进行编辑生成kettle文件时,只能指定一个数据源, 并且我们在开发到发布过程中最少有三套环境,dev, test, prod. 如果按照kettle只能指定一个数据源,对应的kjb/ktr文件就得要三套,首先不说麻烦, 更容易导致出错,测试发到生产的可能.所以我们要解决这个痛点, 就是可以根据我们配置三套环境的数据源,然后根据环境变量自己去更换数据源.

1.踩坑

在查询资料((#^.#)其实就是百度)的过程中,找到如下方法,(以kjb文件为例):
注意该方法为错误方法
踩坑点:在将kjb的路径传到JobMeta里初始化时候就kettle就开始连接数据库了,导致我在测试环境一直报数据库连接错误(作者公司test/dev环境是隔开的本地是没办法连接测试环境)

public static Job executeKjb(String kjbPath, Map<String, String> param) {
        log.info("execute krt, ktrPath={}, param={}", kjbPath, param);
        Job job;
        try {

            KettleEnvironment.init();
            //初始化job路径
            JobMeta jobMeta = new JobMeta(kjbPath, null);
            //替换db链接信息(这里主要设置kjb文件里面的数据库连接信息)
           setDBLinkInfo(KETTLE_DB_PARAM_LIST, jobMeta.getDatabases());
            job = new Job(null, jobMeta);
            //krt文件更换连接信息(修改kjb引用ktr文件里的数据库连接信息)
           Map<JobEntryCopy, JobEntryTrans> activeJobEntryTransformations = job.getActiveJobEntryTransformations();
           activeJobEntryTransformations.entrySet().forEach(jobEntryCopyJobEntryTransEntry -> {
               DatabaseMeta[] connections = jobEntryCopyJobEntryTransEntry.getValue().getUsedDatabaseConnections();
               setDBLinkInfo(KETTLE_DB_PARAM_LIST, Arrays.asList(connections));
            });
            //初始化job参数，脚本中获取参数值：${variableName}
            if (!CollectionUtils.isEmpty(param)) {
                job.injectVariables(param);
            }
            job.start();
            job.waitUntilFinished();
        } catch (Exception e) {
            log.error("kbj文件执行失败", e);
            throw new ServiceException(".kjb文件执行失败", e);
        }
        return job;
    }

//dbParams自己配置的各环境的数据库连接信息
private static void setDBLinkInfo(List<KettleDBParams.KettleDBParam> dbParams, List<DatabaseMeta> databases) {
        //设置DB参数
        if (!CollectionUtils.isEmpty(dbParams)) {
            Map<String, KettleDBParams.KettleDBParam> connNameGroupMap = dbParams.stream()
                    .collect(Collectors.toMap(KettleDBParams.KettleDBParam::getConnName, e -> e, (o, o2) -> o));
            for (DatabaseMeta databaseMeta : databases) {
                KettleDBParams.KettleDBParam dbParam = connNameGroupMap.get(databaseMeta.getName());
                if (dbParam == null) {
                    throw new ServiceException(String.format("执行.ktr失败, 未发现数据库连接【%s】", dbParam.getConnName()));
                }
                log.info("kettle数据库使用的驱动为:{}", databaseMeta.getDriverClass());
                //连接地址
                databaseMeta.setHostname(dbParam.getHost());
                //数据库名称
                databaseMeta.setDBName(dbParam.getDbName());
                //端口
                databaseMeta.setDBPort(dbParam.getPort());
                //用户
                databaseMeta.setUsername(dbParam.getUsername());
                //密码
                databaseMeta.setPassword(dbParam.getPassword());
            }
        } else {
            throw new ServiceException("需要替换的数据库参数为空!!!");
        }
    }

2.正确修改方式(可能是之一)

作者用的是JNDI方式连接数据源

图片.png
配置文件:
kettle解压目录/simple-jndi/jdbc.properties
以下是配置文件内原本内容
/前面就是JNDI名称,上图红框内容, 按照自己需要配置进行配置即可

SampleData/type=javax.sql.DataSource
SampleData/driver=org.h2.Driver
SampleData/url=jdbc:h2:file:samples/db/sampledb;IFEXISTS=TRUE
SampleData/user=PENTAHO_USER
SampleData/password=PASSWORD
SampleDataAdmin/type=javax.sql.DataSource
SampleDataAdmin/driver=org.h2.Driver
SampleDataAdmin/url=jdbc:h2:file:samples/db/sampledb;IFEXISTS=TRUE
SampleDataAdmin/user=PENTAHO_ADMIN
SampleDataAdmin/password=PASSWORD
Quartz/type=javax.sql.DataSource
Quartz/driver=org.hsqldb.jdbcDriver
Quartz/url=jdbc:hsqldb:hsql://localhost/quartz
Quartz/user=pentaho_user
Quartz/password=password
Hibernate/type=javax.sql.DataSource
Hibernate/driver=org.hsqldb.jdbcDriver
Hibernate/url=jdbc:hsqldb:hsql://localhost/hibernate
Hibernate/user=hibuser
Hibernate/password=password
Shark/type=javax.sql.DataSource
Shark/driver=org.hsqldb.jdbcDriver
Shark/url=jdbc:hsqldb:hsql://localhost/shark
Shark/user=sa
Shark/password=
PDI_Operations_Mart/type=javax.sql.DataSource
PDI_Operations_Mart/driver=org.postgresql.Driver
PDI_Operations_Mart/url=jdbc:postgresql://localhost:5432/hibernate?searchpath=pentaho_operations_mart
PDI_Operations_Mart/user=hibuser
PDI_Operations_Mart/password=password
live_logging_info/type=javax.sql.DataSource
live_logging_info/driver=org.postgresql.Driver
live_logging_info/url=jdbc:postgresql://localhost:5432/hibernate?searchpath=pentaho_dilogs
live_logging_info/user=hibuser
live_logging_info/password=password

3.正确方式-java项目中运用

我们只需要根据dev/test/prod各个环境去读取相应的jdbc.properties文件即可
然后我们在知道kettle是如何读取到jdbc.properties文件,
源码如下:

//在init的时候就会设置读取
KettleEnvironment.init();
↓ 一直点下去init方法
public static void init(List<PluginTypeInterface> pluginClasses, boolean simpleJndi) throws KettleException {
        SettableFuture ready;
        if (initialized.compareAndSet((Object)null, ready = SettableFuture.create())) {
            System.setProperties(ConcurrentMapProperties.convertProperties(System.getProperties()));

            try {
                if (!KettleClientEnvironment.isInitialized()) {
                    KettleClientEnvironment.init();
                }
                //这里是加载JNDI方式的方法
                if (simpleJndi) {
                    JndiUtil.initJNDI();
                }

                pluginClasses.forEach(PluginRegistry::addPluginType);
                PluginRegistry.init();
                KettleVariablesList.init();
                initLifecycleListeners();
                ready.set(true);
            } catch (Throwable var5) {
                ready.setException(var5);
                throw var5 instanceof KettleException ? (KettleException)var5 : new KettleException(var5);
            }
        } else {
            ready = (SettableFuture)initialized.get();

            try {
                ready.get();
            } catch (Throwable var4) {
                throw new KettleException(var4);
            }
        }

    }
↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓点进 JndiUtil.initJNDI();
 public static void initJNDI() throws KettleException {
//加载jdbc.properties的路径是从Const.JNDI_DIRECTORY静态变量里获取, 如果为空就加载kettle文件内数据源
        String path = Const.JNDI_DIRECTORY;
        if (path == null || path.equals("")) {
            try {
                File file = new File("simple-jndi");
                path = file.getCanonicalPath();
            } catch (Exception var2) {
                throw new KettleException("Error initializing JNDI", var2);
            }

            Const.JNDI_DIRECTORY = path;
        }

        System.setProperty("java.naming.factory.initial", "org.osjava.sj.SimpleContextFactory");
        System.setProperty("org.osjava.sj.root", path);
        System.setProperty("org.osjava.sj.delimiter", "/");
    }

那我们的思路就来了, 我们在调用kettle文件之前赋值给Const.JNDI_DIRECTORY静态变量我们自己的路径就好了,那么下面就是我自己的加载方式
注意jdbc.properties文件名不要改, 应该默认读取的,作者没有往深里看

import com.xxxx.actuator.exception.ServiceException;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.Const;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.File;

/**
 * 初始化kettle读取数据源配置文件地址
 * 我是把jdbc.properties配置放到服务器上的,springboot打成jar读取文件太麻烦了,
 *以下只是给一个启发, 具体如何读取按照自己项目需求.
 */
@Component
@Slf4j
public class JndiInit {
    @Value("${spring.profiles}")
    private String evn;

    @PostConstruct
    public void init() {
        try {
            String fileSeparator = System.getProperty("file.separator");
            //开发环境获取根目录下文件路径
            if ("dev".equals(evn)) {
                Const.JNDI_DIRECTORY = new File("").getCanonicalPath() + fileSeparator + "jndi" + fileSeparator + evn;

            } else if ("test".equals(evn) || "prod".equals(evn)) {
                //测试/生产环境
                Const.JNDI_DIRECTORY = "/home/service/app/executor/tmp" + fileSeparator + "jndi" + fileSeparator + evn;
            }
        } catch (Exception e) {
            throw new ServiceException("初始化jndi地址失败", e);
        }
    }
}