Java集成kettle开发-解决kettle无法插入空字符串的问题

蔚落 2023-02-20 02:27 5阅读 0赞

# 问题背景 #

今天使用kettle做数据归档时，发现kettle无法插入空字符串，它默认会把空字符串变为null。这就导致问题出现。

# 查询资料 #

通过查资料发现了网上的一个解决办法，首先感谢 [关于kettle的空字符串和NULL的问题][kettle_NULL] 这篇文章。但是他那个是直接使用的kettle客户端，并没有和java集成。  
然后我就去研究源码，看kettle是否支持设置这样的环境变量，我以KettleEnvironment.init();EnvUtil.environmentInit(); 这两个初始化方法为入口，经过仔细研究并没有找到对应的办法；  
无奈之下，我就想：是否可以利用提前生成kettle.properties文件的办法来解决。遂有本文。

# 解决方案 #

直接上代码（本文是在[quartz + kettle8二次开发-实现集群高可用][quartz _ kettle8_-]）基础上继续做的，修改了里面的kettleUtil：

**- 注意看里面的dealWriteNull()方法**

package cc.xxxx.utils;
    
    import cc.xxxx.configs.EnvConfig;
    import lombok.NonNull;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.pentaho.di.core.Const;
    import org.pentaho.di.core.KettleClientEnvironment;
    import org.pentaho.di.core.KettleEnvironment;
    import org.pentaho.di.core.util.EnvUtil;
    import org.pentaho.di.job.Job;
    import org.pentaho.di.job.JobMeta;
    import org.pentaho.di.trans.Trans;
    import org.pentaho.di.trans.TransMeta;
    import org.springframework.stereotype.Component;
    
    import java.io.BufferedWriter;
    import java.io.ByteArrayInputStream;
    import java.io.File;
    import java.io.FileWriter;
    import java.util.Map;
    
    /** * kettle工具类 * 使用静态块的原因： * 1.执行ktr和kjb都要执行初始化，为了代码的复用； * 2.由于quartz的定时任务是异步的，在springBoot刚刚启动完成就可能会执行ktr和kjb，所以kettle的初始化工作 * 必须在springBoot启动完成前执行，故使用静态块； * @author wangll */
    @Slf4j
    @Component
    @RequiredArgsConstructor
    public class KettleUtil { 
    
        @NonNull
        private EnvConfig envConfig;
    
        static { 
            log.info("正在初始化kettle ...");
            try { 
                dealWriteNull();
                KettleEnvironment.init();
                EnvUtil.environmentInit();
            }catch (Exception e){ 
                log.error("初始化kettle失败",e);
                System.exit(0);
            }
            log.info("初始化kettle成功！");
        }
    
        /** * 执行ktr文件内容 * @param file * @param beginTime * @param endTime */
        public void runTrans(@NonNull String file, String beginTime, String endTime){ 
            log.info("正在执行ktr ...");
            try { 
                TransMeta transMeta = new TransMeta(new ByteArrayInputStream(file.getBytes("UTF-8")), null, true, null, null);
                Trans trans = new Trans(transMeta);
                Map<String,String> params = this.dealParams(beginTime,endTime);
                params.forEach((k,v)->trans.setVariable(k,v));
                trans.prepareExecution(null);
                trans.startThreads();
                trans.waitUntilFinished();
                if (trans.getErrors() != 0) { 
                    log.error("执行ktr过程中存在错误");
                }
            }catch (Exception e){ 
                log.error("runTrans失败",e);
            }
            log.info("ktr执行完毕！");
        }
    
        /** * 执行kjb文件内容 * @param file * @param beginTime * @param endTime */
        public void runJob(@NonNull String file, String beginTime, String endTime){ 
            log.info("正在执行kjb ...");
            try { 
                JobMeta jobMeta = new JobMeta(new ByteArrayInputStream(file.getBytes("UTF-8")), null, null);
                Job job = new Job(null, jobMeta);
                Map<String,String> params = this.dealParams(beginTime,endTime);
                params.forEach((k,v)->job.setVariable(k,v));
                job.start();
                job.waitUntilFinished();
                if (job.getErrors() != 0) { 
                    log.error("执行kjb过程中存在错误");
                }
            }catch (Exception e){ 
                log.error("runJob失败",e);
            }
            log.info("kjb执行完毕...");
        }
    
        /** * 处理公共参数 */
        private Map<String,String> dealParams(String beginTime, String endTime){ 
            Map<String, String> params = envConfig.getDamEnv();
            params.put("beginTime",beginTime);
            params.put("endTime",endTime);
            if (params.containsKey("beginTime")){ 
                params.put("beginTimeId",TimeUtil.date2timeId(params.get("beginTime")));
            }
            if (params.containsKey("endTime")){ 
                params.put("endTimeId",TimeUtil.date2timeId(params.get("endTime")));
            }
            log.debug("kettle参数：{}",params);
            return params;
        }
    
        /** * 解决kettle无法写入空字符串的问题 * window环境中，需要在C:\Users\wangll\.kettle\kettle.properties中写入如下配置； * linux环境中，需要在/root/.kettle/kettle.properties中写入如下配置。 * 故为了方便直接使用它自带的方法去生成上述文件 */
        public static void dealWriteNull()throws Exception{ 
            String directory = Const.getKettleDirectory();
            File dir = new File(directory);
            dir.mkdirs();
            KettleClientEnvironment.createKettleHome();
            String kpFile = directory + Const.FILE_SEPARATOR + "kettle.properties";
            File file = new File(kpFile);
            FileWriter fw = new FileWriter(file);
            BufferedWriter bw=new BufferedWriter(fw);
            bw.write("KETTLE_EMPTY_STRING_DIFFERS_FROM_NULL=Y");
            bw.close();
            fw.close();
        }
    }

[kettle_NULL]: https://blog.csdn.net/lost_wh/article/details/53466787
[quartz _ kettle8_-]: https://blog.csdn.net/tiandixuanwuliang/article/details/97968834