Springboot整合ETL引擎Kettle的使用

Dear 丶 2023-10-12 08:41 20阅读 0赞

### 简介 ###

**ETL**是英文Extract-Transform-Load的缩写，用来描述将数据从源端经过抽取(extract)、转换(transform)、加载(load)至目的端的过程，它能够对各种分布的、异构的源数据(如关系数据)进行抽取，按照预先设计的规则将不完整数据、重复数据以及错误数据等“脏"数据内容进行清洗,得到符合要求的“干净”数据,并加载到[数据仓库][Link 1]中进行存储,这些“干净”数据就成为了数据分析、数据挖掘的基石。

**kettle**是一个开源ETL工具。kettle提供了基于java的图形化界面，使用很方便。kettle提供了基于 JAVA的脚步编写功能，可以灵活地自定义ETL过程，使自行定制、批量处理等成为可能，这才是一个程序员需要做的工作，而不仅是象使用word一样操作 kettle用户界面。

![fb67765651814ee38dd50e71725900bd.png][]

### 环境集成： ###

参考：[java集成kettle教程（附示例代码）\_kettle java\_成伟平2022的博客-CSDN博客][java_kettle_kettle java_2022_-CSDN]

### 代码： ###

pom.xml添加：

<dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid</artifactId>
                <version>1.2.11</version>
            </dependency>
    
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>kettle-core</artifactId>
                <version>8.2.0.7-719</version>
                <scope>system</scope>
                <systemPath>${project.basedir}/lib/kettle-core-8.2.0.7-719.jar</systemPath>
            </dependency>
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>kettle-engine</artifactId>
                <version>8.2.0.7-719</version>
                <scope>system</scope>
                <systemPath>${project.basedir}/lib/kettle-engine-8.2.0.7-719.jar</systemPath>
            </dependency>
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>metastore</artifactId>
                <version>8.2.0.7-719</version>
                <scope>system</scope>
                <systemPath>${project.basedir}/lib/metastore-8.2.0.7-719.jar</systemPath>
            </dependency>
            
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-vfs2</artifactId>
                <version>2.2</version>
            </dependency>
            <dependency>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
                <version>17.0</version>
            </dependency>
            <dependency>
                <groupId>commons-io</groupId>
                <artifactId>commons-io</artifactId>
                <version>2.2</version>
            </dependency>
            <dependency>
                <groupId>commons-lang</groupId>
                <artifactId>commons-lang</artifactId>
                <version>2.6</version>
            </dependency>
            <dependency>
                <groupId>commons-codec</groupId>
                <artifactId>commons-codec</artifactId>
                <version>1.10</version>
            </dependency>
            <dependency>
                <groupId>com.jcraft</groupId>
                <artifactId>jsch</artifactId>
                <version>0.1.54</version>
            </dependency>
            <dependency>
                <groupId>net.sourceforge.jexcelapi</groupId>
                <artifactId>jxl</artifactId>
                <version>2.6.12</version>
            </dependency>

@RestController
    @RequestMapping("${application.admin-path}/etl-kettl")
    //@Api(tags = "ETL-Kettle的demo接口")
    public class KettleDemoContrllor {
    	@Resource
    	KettleService kettleService;
    
    	@GetMapping("/execKtr")
    	//@ApiOperation("执行ktr文件")
    	private Object runKtr(String filename) throws Exception {
    		return R.buildOkData(kettleService.runTaskKtr(filename,null).toString());
    	}
    
    	@GetMapping("/execKjb")
    	//@ApiOperation("执行kjb文件")
    	private Object runKjb(String filename) throws Exception {
    		return R.buildOkData(kettleService.runTaskKjb(filename, null).toString());
    	}
    }

public interface KettleService {
        /**
         * 开始执行ETL任务（ktr文件）
         *
         * @param taskFileName 执行的任务文件名（ktr）
         * @param params 执行任务输入的参数
         * @return 运行结果
         * @throws Exception 没有找到配置文件，Kettle的运行异常不会抛出
         */
        Object runTaskKtr(String taskFileName, Map<String, String> params) throws Exception;
        /**
         * 开始执行ETL任务（kjb文件）
         *
         * @param taskFileName 执行的任务文件名（kjb）
         * @param params 执行任务输入的参数
         * @return 运行结果
         * @throws Exception 没有找到配置文件，Kettle的运行异常不会抛出
         */
        Object runTaskKjb(String taskFileName, Map<String, String> params) throws Exception;
    }

@Service
    public class KettleServiceImpl implements KettleService {
    
        @Value("${kettle.script.path}")
        private String kettleScriptPath;
    
        private static final Logger logger = LoggerFactory.getLogger("kettle-service-log");
    
        private final List<KtrMeta> KTR_METAS = new ArrayList<>();
        private final List<KjbMeta> KJB_METAS = new ArrayList<>();
    
        private List<String> getFiles(String path, String subName) {
            List<String> files = new ArrayList<>();
            File file = new File(path);
            File[] tempList = file.listFiles();
            if (tempList == null){
                return files;
            }
            for (File value : tempList) {
                if (value.isFile()) {
                    if (Objects.equals(value.toString().substring(value.toString().length() - 3), subName)) {
                        files.add(value.getName());
                    }
                }
            }
            return files;
        }
    
        //采用单列模式，项目启动时加载环境，加载所有的转换配置、任务配置，后续执行就会快一点
        //@PostConstruct
        public void init() throws KettleException {
            logger.info("----------------------开始初始化ETL配置------------------------");
            KettleEnvironment.init();
            List<String> ktrFiles = getFiles(kettleScriptPath, "ktr");
            List<String> kjbFiles = getFiles(kettleScriptPath, "kjb");
            logger.info("需要加载的转换为：" + ktrFiles.toString());
            logger.info("需要加载的任务为：" + kjbFiles.toString());
            logger.info("----------------------开始加载ETL配置--------------------------");
            for (String ktrFile : ktrFiles) {
                KtrMeta ktrMeta = new KtrMeta();
                ktrMeta.setName(ktrFile);
                ktrMeta.setTransMeta(new TransMeta(kettleScriptPath + ktrFile));
                KTR_METAS.add(ktrMeta);
                logger.info("成功加载转换配置：" + ktrFile);
            }
            for (String kjbFile : kjbFiles) {
                KjbMeta kjbMeta = new KjbMeta();
                kjbMeta.setName(kjbFile);
                kjbMeta.setJobMeta(new JobMeta(kettleScriptPath + kjbFile, null));
                KJB_METAS.add(kjbMeta);
                logger.info("成功加载任务配置：" + kjbFile);
            }
            logger.info("----------------------全部ETL配置加载完毕-----------------------");
        }
    
    
        @Override
        public Object runTaskKtr(String ktrFileName, Map<String, String> params) {
            logger.info("开始执行转换：" + ktrFileName);
            TransMeta transMeta = null;
            for (KtrMeta ktrMeta : KTR_METAS) {
                if(Objects.equals(ktrFileName,ktrMeta.getName())){
                    transMeta = ktrMeta.getTransMeta();
                    break;
                }
            }
            //如果在缓存的列表里面没找到需要自信的配置，尝试手动加载
            try {
                if (transMeta == null) {
                    logger.warn("资源池没有找到配置文件：" + ktrFileName+"  尝试二次加载！");
                    KettleEnvironment.init();
                    transMeta = new TransMeta(kettleScriptPath + File.separator + ktrFileName);
                    if(transMeta==null) throw new RuntimeException("未找到需要执行的转换配置文件：");
                }
                Trans trans = new Trans(transMeta);
                if (params != null) {
                    for (Map.Entry<String, String> entry : params.entrySet()) {
                        trans.setParameterValue(entry.getKey(), entry.getValue());
                    }
                }
                //trans.prepareExecution(null);
                //trans.startThreads(); //启用新的线程加载
                trans.execute(null);
                trans.waitUntilFinished();
                return trans.getResult();
            }catch (Exception e)
            {
                e.printStackTrace();
                return e.getMessage();
            }
    
        }
    
        @Override
        public Object runTaskKjb(String objFileName, Map<String, String> params) throws Exception {
            logger.info("开始执行任务：" + objFileName);
            JobMeta jobMeta = null;
            for (KjbMeta kjbMeta : KJB_METAS) {
                if(Objects.equals(objFileName,kjbMeta.getName())){
                    jobMeta = kjbMeta.getJobMeta();
                }
            }
            try {
                if (jobMeta == null) {
                    logger.warn("资源池没有找到配置文件：" + objFileName+"  尝试二次加载！");
                    KettleEnvironment.init();
                    jobMeta = new JobMeta(kettleScriptPath + File.separator + objFileName,null);
                    if(jobMeta==null) throw new RuntimeException("未找到需要执行的任务配置文件："+objFileName);
                }
                Job job = new Job(null, jobMeta);
                if (params != null) {
                    for (Map.Entry<String, String> entry : params.entrySet()) {
                        job.setParameterValue(entry.getKey(), entry.getValue());
                    }
                }
                job.start();
                job.waitUntilFinished();
                return job.getResult();
            }catch (Exception e)
            {
                e.printStackTrace();
                return e.getMessage();
            }
    
        }
    }

@Data
    public class KtrMeta {
    	private TransMeta transMeta;
    	private String name;
    }

@Data
    public class KjbMeta {
    	private JobMeta jobMeta;
    	private String name;
    }

### 总结： ###

集成后感觉没什么必要集成到项目里面去。关键还是需要学会工具的使用，以便进行数据收集与治理。

参考：[1\_ETL和Kettle概述\_哔哩哔哩\_bilibili][1_ETL_Kettle_bilibili]

[下载： kettle工具下载][kettle]

[Link 1]: https://so.csdn.net/so/search?q=%E6%95%B0%E6%8D%AE%E4%BB%93%E5%BA%93&spm=1001.2101.3001.7020
[fb67765651814ee38dd50e71725900bd.png]: https://img-blog.csdnimg.cn/fb67765651814ee38dd50e71725900bd.png
[java_kettle_kettle java_2022_-CSDN]: https://blog.csdn.net/pingweicheng/article/details/103547943
[1_ETL_Kettle_bilibili]: https://www.bilibili.com/video/BV1NT4y1c7o8?p=1&vd_source=af6a2c7d4600ce9881d1d1f7c82cc29a
[kettle]: https://s3.amazonaws.com/kettle-neo4j/kettle-neo4j-remix-8.2.0.7-719-REMIX.zip