quartz + kettle8二次开发-实现集群高可用

Bertha 。 2021-11-14 13:46 1366阅读 0赞

### 文章目录 ###

*   *  背景说明
     *  项目结构
     *  开发步骤
     *   *  1、引入quartz依赖，和kettle依赖
         *  2、配置quartz
         *  3、编写系统初始化类，在程序启动时，添加quartz定时任务
         *  4、编写quartz任务执行类，这个类中会去调用kettle的方法，从而执行kettle脚本
         *  5、其他类
     *  效果演示

## 背景说明 ##

*  1、etl是报表开发/大数据开发的必备步骤，而其中免费、开源、且最好用的工具就是kettle。关于kettle（pdi-ce）可视化界面工具的使用和介绍请大家自行百度。本文主要介绍如何把kettle集成到我们的项目中，来进行二次开发。
 *  2、通过查询资料发现：大多数公司的做法是部署一套kettle集群（伪集群），然后使用jenkins去定时调度kettle集群。甚至有公司直接用linux中的cron来做定时任务调度的工作。但是这里存在两个问题：
    
     *  **第一**：kettle集群为“主-从”式的伪集群，存在单点故障问题，如下图：  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center]  
        （当然网上也有很多解决办法，例如，master可以使用keeplive做成双活来避免单点问题，但是需要多引入一个服务。增加维护成本）
     *  **第二**：定时任务本身也存在单点故障问题。
        
        ![在这里插入图片描述][20191014170427325.png_pic_center]  
        （这个问题网上也有解决办法，例如使用第三方开源的任务调度系统，例如kettle-manager等，其实km的底层调度仍然用的quartz，也就是本文准备介绍的。km拥有界面化操作，能够直接在界面上对定时job进行crud，个人觉得还是很方便的。感兴趣的可以自行研究。）
 *  3、本文介绍的quartz+kettle架构  
    \*\*本项目的做法是：\*\*直接把kettle集成到程序中，然后使用quartz来调度kettle。不再需要依赖已有的kettle集群，并且支持服务的高可用。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 1]  
（**注意**：由于不再依赖kettle集群，故没有master/slave的区别，也没有master分发任务的动作。所有任务在同一时刻只会在一个实例上做，会造成实例压力大的问题。处理任务的效率可能会低于伪集群式）

## 项目结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 2]

## 开发步骤 ##

*  项目的核心逻辑是：  
    1、在springboot项目启动时，会最先执行InitRunner类，这个类会把quartz的定时任务写入到数据库中；  
    2、当定时时间到时，会去执行Job类，这个类会去调用kettle的方法，从而执行kettle脚本。
 *  项目开发步骤是：  
    ![在这里插入图片描述][20191014173220445.png_pic_center]

### 1、引入quartz依赖，和kettle依赖 ###

*  注意其中的kettle三个核心库

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.6.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.wllfengshu</groupId>
        <artifactId>etl-kettle</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>etl-kettle</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
            <kettle.version>8.2.0.0-342</kettle.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-aop</artifactId>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.45</version>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.54</version>
            </dependency>
            <dependency>
                <groupId>net.logstash.logback</groupId>
                <artifactId>logstash-logback-encoder</artifactId>
                <version>5.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-vfs2</artifactId>
                <version>2.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
                <version>3.8.1</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.11</version>
            </dependency>
            <dependency>
                <groupId>commons-io</groupId>
                <artifactId>commons-io</artifactId>
                <version>2.6</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.4</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-quartz</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-batch</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-thymeleaf</artifactId>
            </dependency>
            <dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>1.3.2</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid-spring-boot-starter</artifactId>
                <version>1.1.10</version>
            </dependency>
            <dependency>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
                <version>20.0</version>
            </dependency>
            
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>kettle-core</artifactId>
                <version>${kettle.version}</version>
            </dependency>
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>kettle-engine</artifactId>
                <version>${kettle.version}</version>
            </dependency>
            <dependency>
                <groupId>pentaho-kettle</groupId>
                <artifactId>metastore</artifactId>
                <version>${kettle.version}</version>
            </dependency>
            
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-jar-plugin</artifactId>
                    <configuration>
                        <archive>
                            <manifest>
                                <addClasspath>true</addClasspath>
                                <useUniqueVersions>false</useUniqueVersions>
                                <classpathPrefix>lib/</classpathPrefix>
                                <mainClass>com.wllfengshu.etlkettle.EtlKettleApplication</mainClass>
                            </manifest>
                        </archive>
                        <excludes>
                            <exclude>/workspace/</exclude>
                        </excludes>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    </project>

### 2、配置quartz ###

server:
      port: 8080
      servlet:
        context-path: /etl/1.0
    spring:
      application:
        name: etl
      datasource:
        url: ${ db_url}
        username: ${ db_username}
        password: ${ db_password}
        driver-class-name: com.mysql.jdbc.Driver
        type: com.alibaba.druid.pool.DruidDataSource
      quartz:
        job-store-type: jdbc
        jdbc:
          initialize-schema: always
        properties:
          org:
            quartz:
              scheduler:
                instanceName: QuartzScheduler
                instanceId: CLUSTERED
              jobStore:
                class: org.quartz.impl.jdbcjobstore.JobStoreTX
                driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
                tablePrefix: ETL_QRTZ_
                isClustered: true
                clusterCheckinInterval: 10000
                useProperties: false
              threadPool:
                class: org.quartz.simpl.SimpleThreadPool
                threadCount: 50
                threadPriority: 5
                threadsInheritContextClassLoaderOfInitializingThread: true
    logging:
      level:
        cc.wellcloud.etl.dao : debug

### 3、编写系统初始化类，在程序启动时，添加quartz定时任务 ###

package com.wllfengshu.etlkettle.init;
    
    import com.wllfengshu.etlkettle.common.Constant;
    import com.wllfengshu.etlkettle.model.QuartzJob;
    import com.wllfengshu.etlkettle.utils.QuartzUtil;
    import lombok.NonNull;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.boot.CommandLineRunner;
    import org.springframework.stereotype.Component;
    
    /** * 系统初始化 * @author wangll */
    @Slf4j
    @Component
    @RequiredArgsConstructor
    public class InitRunner implements CommandLineRunner { 
    
        private static final String CRON_MINUTE = "* */2 * * * ?";
        private static final String JOB_NAME = "etl_minute";
        private static final String TRIGGER_NAME = "etl_minute";
    
        @NonNull
        private QuartzUtil quartzUtil;
    
        @Override
        public void run(String... args){ 
            log.info("正在初始化全部定时任务...");
            QuartzJob m = new QuartzJob();
            m.setJobName(JOB_NAME);
            m.setTriggerName(TRIGGER_NAME);
            m.setCronExpression(CRON_MINUTE);
            m.setDescription("etl 2分钟的定时任务");
            quartzUtil.addJob(m,Constant.JOB_CLASS_NAME);
            log.info("定时任务添加完毕");
        }
    }

### 4、编写quartz任务执行类，这个类中会去调用kettle的方法，从而执行kettle脚本 ###

package com.wllfengshu.etlkettle.job;
    
    import com.wllfengshu.etlkettle.common.Constant;
    import com.wllfengshu.etlkettle.utils.DbUtil;
    import com.wllfengshu.etlkettle.utils.KettleUtil;
    import lombok.NonNull;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.quartz.JobExecutionContext;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.scheduling.quartz.QuartzJobBean;
    import org.springframework.stereotype.Component;
    
    import java.io.File;
    import java.util.HashMap;
    import java.util.Map;
    
    /** * 定时任务 * @author wangll */
    @Slf4j
    @Component
    @RequiredArgsConstructor
    public class Job extends QuartzJobBean { 
    
        @NonNull
        private KettleUtil kettleUtil;
    
        @Value("${db_url}")
        private String dbUrl;
        @Value("${db_username}")
        private String dbUsername;
        @Value("${db_password}")
        private String dbPassword;
    
        @Override
        protected void executeInternal(JobExecutionContext jobExecutionContext){ 
            //1 设置kettle参数(把程序中的参数传入到kettle中)
            Map<String,String> params = new HashMap<>(16);
            //1.1 设置数据库信息
            Map<String, String> mapInput = DbUtil.dealDb(dbUrl, dbUsername, dbPassword);
            params.put("dbIp",mapInput.get("dbIp"));
            params.put("dbPort",mapInput.get("dbPort"));
            params.put("dbName",mapInput.get("dbName"));
            params.put("dbUsername",mapInput.get("dbUsername"));
            params.put("dbPassword",mapInput.get("dbPassword"));
            log.info("kettle配置的参数：{}",params);
            //2 取所有kettle脚本
            File workspace = new File(Constant.WORKSPACE_PATH);
            File[] files = workspace.listFiles();
            //2.1 执行kettle
            for (File file:files){ 
                String fileName = file.getName();
                try { 
                    if (fileName.endsWith(Constant.FILE_SUFFIX_KTR)){ 
                        kettleUtil.runTrans(file, params);
                    }else if (fileName.endsWith(Constant.FILE_SUFFIX_KJB)){ 
                        kettleUtil.runJob(file, params);
                    }else { 
                        log.error("文件名不合法,它必须是ktr或kjb");
                        continue;
                    }
                } catch (Exception e) { 
                    log.error("执行ktr或kjb失败,fileName={}",fileName,e);
                }
            }
        }
    }

### 5、其他类 ###

//(1) 常量类
    package com.wllfengshu.etlkettle.common;
    
    import lombok.extern.slf4j.Slf4j;
    
    /** * 公共常量集合 * * @author wangll */
    @Slf4j
    public class Constant { 
    
        /** * ktr文件后缀 */
        public static final String FILE_SUFFIX_KTR = "ktr";
    
        /** * kjb文件后缀 */
        public static final String FILE_SUFFIX_KJB = "kjb";
    
        /** * job组 */
        public static final String JOB_GROUP = "ETL";
    
        /** * job的执行类 */
        public static final String JOB_CLASS_NAME = "com.wllfengshu.etlkettle.job.Job";
    
        /** * ktr和kjb文件存放路径 */
        public static final String WORKSPACE_PATH = "/home/listen/Apps/workspace";
    
    }
    
    //(2) quartz的实体类
    package com.wllfengshu.etlkettle.model;
    
    import lombok.Data;
    
    import java.io.Serializable;
    
    /** * job实体类 * @author wangll */
    @Data
    public class QuartzJob implements Serializable { 
    
        private static final long serialVersionUID = 1L;
    
        /** * 任务名称 */
        private String jobName;
    
        /** * 任务分组 */
        private String jobGroup;
    
        /** * 任务描述 */
        private String description;
    
        /** * 执行类 */
        private String jobClassName;
    
        /** * 执行时间 */
        private String cronExpression;
    
        /** * 触发器名 */
        private String triggerName;
    
        /** * 触发器状态 */
        private String triggerState;
    
    }
    
    //(3) kettle执行类
    package com.wllfengshu.etlkettle.utils;
    
    import lombok.NonNull;
    import lombok.extern.slf4j.Slf4j;
    import org.pentaho.di.core.KettleEnvironment;
    import org.pentaho.di.core.exception.KettleException;
    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.File;
    import java.io.FileInputStream;
    import java.io.FileNotFoundException;
    import java.util.Map;
    
    /** * kettle工具类 * 使用静态块的原因： * 1.执行ktr和kjb都要执行初始化，为了代码的复用； * 2.由于quartz的定时任务是异步的，在springBoot刚刚启动完成就可能会执行ktr和kjb，所以kettle的初始化工作 * 必须在springBoot启动完成前执行，故使用静态块； * @author wangll */
    @Slf4j
    @Component
    public class KettleUtil { 
    
        static { 
            log.info("正在初始化kettle ...");
            try { 
                KettleEnvironment.init();
                EnvUtil.environmentInit();
            }catch (Exception e){ 
                log.error("初始化kettle失败",e);
                System.exit(0);
            }
            log.info("初始化kettle成功！");
        }
    
        /** * 执行ktr文件 * @param file * @param params */
        public void runTrans(@NonNull File file, Map<String,String> params) throws KettleException, FileNotFoundException { 
            log.info("正在执行ktr ...");
            TransMeta transMeta = new TransMeta(new FileInputStream(file), null, true, null, null);
            Trans trans = new Trans(transMeta);
            params.forEach((k,v)->
                    trans.setVariable(k,v)
            );
            trans.prepareExecution(null);
            trans.startThreads();
            trans.waitUntilFinished();
            if (trans.getErrors() != 0) { 
                log.error("执行ktr过程中存在错误");
            }
            log.info("ktr执行完毕！");
        }
    
        /** * 执行kjb文件 * @param file * @param params */
        public void runJob(@NonNull File file,Map<String,String> params) throws KettleException, FileNotFoundException { 
            log.info("正在执行kjb ...");
            JobMeta jobMeta = new JobMeta(new FileInputStream(file), null, null);
            Job job = new Job(null, jobMeta);
            params.forEach((k,v)->
                    job.setVariable(k,v)
            );
            job.start();
            job.waitUntilFinished();
            if (job.getErrors() != 0) { 
                log.error("执行kjb过程中存在错误");
            }
            log.info("kjb执行完毕...");
        }
    }
    
    
    //(4) quartz工具类
    package com.wllfengshu.etlkettle.utils;
    
    import com.wllfengshu.etlkettle.common.Constant;
    import com.wllfengshu.etlkettle.model.QuartzJob;
    import lombok.NonNull;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.quartz.*;
    import org.springframework.stereotype.Component;
    
    /** * quartz工具类 * @author wangll */
    @Slf4j
    @Component
    @RequiredArgsConstructor
    public class QuartzUtil { 
    
        @NonNull
        private Scheduler scheduler;
    
        /** * 添加任务 */
        public void addJob(@NonNull QuartzJob quartz, @NonNull String jobClassName){ 
            log.info("正在添加定时任务，quartz={} ...",quartz);
            try { 
                //1 构建jobKey判断任务是否已经存在
                JobKey jobKey = new JobKey(quartz.getJobName(),Constant.JOB_GROUP);
                if (scheduler.checkExists(jobKey)) { 
                    log.info("该定时任务已存在！");
                    return;
                }
                //2 构建job信息
                Class cls = Class.forName(jobClassName);
                JobDetail job = JobBuilder.newJob(cls).withIdentity(jobKey)
                        .withDescription(quartz.getDescription()).build();
                //3 触发时间点
                CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(quartz.getCronExpression().trim());
                //4 构建triggerKey判断触发器是否已经存在
                TriggerKey triggerKey = new TriggerKey(quartz.getTriggerName(),Constant.JOB_GROUP);
                if (scheduler.checkExists(triggerKey)) { 
                    log.info("该触发器已存在！");
                    return;
                }
                Trigger trigger = TriggerBuilder.newTrigger().withIdentity(triggerKey)
                        .startNow().withSchedule(cronScheduleBuilder).build();
                //5 交由Scheduler安排触发
                scheduler.scheduleJob(job, trigger);
            } catch (Exception e) { 
                log.error("保存job失败",e);
                // throw CustomException
            }
            log.info("定时任务添加成功！");
        }
    }
    
    //(5) 把连接mysql完整的url分解为ip/port/dbName
    package com.wllfengshu.etlkettle.utils;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /** * 数据库工具类 * @author wangll */
    @Slf4j
    public class DbUtil { 
    
        /** * 处理数据库配置 * @param dbUrl * @param dbUsername * @param dbPassword * @return */
        public static Map<String,String> dealDb(String dbUrl, String dbUsername, String dbPassword) { 
            Map<String,String> map = new HashMap<>(5);
            //1 截取出ip:port/dbName
            String temp = dbUrl.substring("jdbc:mysql://".length(),
                    dbUrl.contains("?") ? dbUrl.indexOf("?") : dbUrl.length());
            //2 分离ip
            String[] ipPortDb = temp.split(":");
            map.put("dbIp",ipPortDb[0]);
            //3 分离port和dbName
            String[] portDb = ipPortDb[1].split("/");
            map.put("dbPort",portDb[0]);
            map.put("dbName",portDb[1]);
            //4 设置用户名和密码
            map.put("dbUsername",dbUsername);
            map.put("dbPassword",dbPassword);
            log.info("数据库信息处理完毕，dbInfo={}",map);
            return map;
        }
    }
    
    //(6) springboot启动类
    package com.wllfengshu.etlkettle;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class EtlKettleApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(EtlKettleApplication.class, args);
        }
    
    }

**注意**：

*  本项目使用了lombok插件，请先确保idea安装了该插件；
 *  本文是把数据库信息作为参数传到kettle脚本中的，其实kettle库中的DatabaseMeta类可以直接设置数据库信息，大家可以自己优化；
 *  本项目是把kettle脚本（ktr和kjb）存放在项目的resources/workspace目录中，然后在docker打包时把所有kettle脚本都移动到/home/listen/Apps/workspace路径下。大家也可以直接把脚本和程序打包到一个jar中，但是需要解决如何读取jar中的文件的问题。

## 效果演示 ##

1、quartz会自动建表，表如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 3]  
2、t\_user表如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 4]  
3、使用脚本插入的效果：

![在这里插入图片描述][20191017144537463.png_pic_center]

4、把编写的kettle脚本放到workspace目录下，例如，insert\_user.ktr脚本是：  
（作用是：向数据库的user表插入一条记录）

<?xml version="1.0" encoding="UTF-8"?>
    <transformation>
      <info>
        <name>插入user</name>
        <description/>
        <extended_description/>
        <trans_version/>
        <trans_type>Normal</trans_type>
        <trans_status>2</trans_status>
        <directory>/</directory>
        <parameters>
        </parameters>
        <log>
          <trans-log-table>
            <connection/>
            <schema/>
            <table/>
            <size_limit_lines/>
            <interval/>
            <timeout_days/>
            <field>
              <id>ID_BATCH</id>
              <enabled>Y</enabled>
              <name>ID_BATCH</name>
            </field>
            <field>
              <id>CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>CHANNEL_ID</name>
            </field>
            <field>
              <id>TRANSNAME</id>
              <enabled>Y</enabled>
              <name>TRANSNAME</name>
            </field>
            <field>
              <id>STATUS</id>
              <enabled>Y</enabled>
              <name>STATUS</name>
            </field>
            <field>
              <id>LINES_READ</id>
              <enabled>Y</enabled>
              <name>LINES_READ</name>
              <subject/>
            </field>
            <field>
              <id>LINES_WRITTEN</id>
              <enabled>Y</enabled>
              <name>LINES_WRITTEN</name>
              <subject/>
            </field>
            <field>
              <id>LINES_UPDATED</id>
              <enabled>Y</enabled>
              <name>LINES_UPDATED</name>
              <subject/>
            </field>
            <field>
              <id>LINES_INPUT</id>
              <enabled>Y</enabled>
              <name>LINES_INPUT</name>
              <subject/>
            </field>
            <field>
              <id>LINES_OUTPUT</id>
              <enabled>Y</enabled>
              <name>LINES_OUTPUT</name>
              <subject/>
            </field>
            <field>
              <id>LINES_REJECTED</id>
              <enabled>Y</enabled>
              <name>LINES_REJECTED</name>
              <subject/>
            </field>
            <field>
              <id>ERRORS</id>
              <enabled>Y</enabled>
              <name>ERRORS</name>
            </field>
            <field>
              <id>STARTDATE</id>
              <enabled>Y</enabled>
              <name>STARTDATE</name>
            </field>
            <field>
              <id>ENDDATE</id>
              <enabled>Y</enabled>
              <name>ENDDATE</name>
            </field>
            <field>
              <id>LOGDATE</id>
              <enabled>Y</enabled>
              <name>LOGDATE</name>
            </field>
            <field>
              <id>DEPDATE</id>
              <enabled>Y</enabled>
              <name>DEPDATE</name>
            </field>
            <field>
              <id>REPLAYDATE</id>
              <enabled>Y</enabled>
              <name>REPLAYDATE</name>
            </field>
            <field>
              <id>LOG_FIELD</id>
              <enabled>Y</enabled>
              <name>LOG_FIELD</name>
            </field>
            <field>
              <id>EXECUTING_SERVER</id>
              <enabled>N</enabled>
              <name>EXECUTING_SERVER</name>
            </field>
            <field>
              <id>EXECUTING_USER</id>
              <enabled>N</enabled>
              <name>EXECUTING_USER</name>
            </field>
            <field>
              <id>CLIENT</id>
              <enabled>N</enabled>
              <name>CLIENT</name>
            </field>
          </trans-log-table>
          <perf-log-table>
            <connection/>
            <schema/>
            <table/>
            <interval/>
            <timeout_days/>
            <field>
              <id>ID_BATCH</id>
              <enabled>Y</enabled>
              <name>ID_BATCH</name>
            </field>
            <field>
              <id>SEQ_NR</id>
              <enabled>Y</enabled>
              <name>SEQ_NR</name>
            </field>
            <field>
              <id>LOGDATE</id>
              <enabled>Y</enabled>
              <name>LOGDATE</name>
            </field>
            <field>
              <id>TRANSNAME</id>
              <enabled>Y</enabled>
              <name>TRANSNAME</name>
            </field>
            <field>
              <id>STEPNAME</id>
              <enabled>Y</enabled>
              <name>STEPNAME</name>
            </field>
            <field>
              <id>STEP_COPY</id>
              <enabled>Y</enabled>
              <name>STEP_COPY</name>
            </field>
            <field>
              <id>LINES_READ</id>
              <enabled>Y</enabled>
              <name>LINES_READ</name>
            </field>
            <field>
              <id>LINES_WRITTEN</id>
              <enabled>Y</enabled>
              <name>LINES_WRITTEN</name>
            </field>
            <field>
              <id>LINES_UPDATED</id>
              <enabled>Y</enabled>
              <name>LINES_UPDATED</name>
            </field>
            <field>
              <id>LINES_INPUT</id>
              <enabled>Y</enabled>
              <name>LINES_INPUT</name>
            </field>
            <field>
              <id>LINES_OUTPUT</id>
              <enabled>Y</enabled>
              <name>LINES_OUTPUT</name>
            </field>
            <field>
              <id>LINES_REJECTED</id>
              <enabled>Y</enabled>
              <name>LINES_REJECTED</name>
            </field>
            <field>
              <id>ERRORS</id>
              <enabled>Y</enabled>
              <name>ERRORS</name>
            </field>
            <field>
              <id>INPUT_BUFFER_ROWS</id>
              <enabled>Y</enabled>
              <name>INPUT_BUFFER_ROWS</name>
            </field>
            <field>
              <id>OUTPUT_BUFFER_ROWS</id>
              <enabled>Y</enabled>
              <name>OUTPUT_BUFFER_ROWS</name>
            </field>
          </perf-log-table>
          <channel-log-table>
            <connection/>
            <schema/>
            <table/>
            <timeout_days/>
            <field>
              <id>ID_BATCH</id>
              <enabled>Y</enabled>
              <name>ID_BATCH</name>
            </field>
            <field>
              <id>CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>CHANNEL_ID</name>
            </field>
            <field>
              <id>LOG_DATE</id>
              <enabled>Y</enabled>
              <name>LOG_DATE</name>
            </field>
            <field>
              <id>LOGGING_OBJECT_TYPE</id>
              <enabled>Y</enabled>
              <name>LOGGING_OBJECT_TYPE</name>
            </field>
            <field>
              <id>OBJECT_NAME</id>
              <enabled>Y</enabled>
              <name>OBJECT_NAME</name>
            </field>
            <field>
              <id>OBJECT_COPY</id>
              <enabled>Y</enabled>
              <name>OBJECT_COPY</name>
            </field>
            <field>
              <id>REPOSITORY_DIRECTORY</id>
              <enabled>Y</enabled>
              <name>REPOSITORY_DIRECTORY</name>
            </field>
            <field>
              <id>FILENAME</id>
              <enabled>Y</enabled>
              <name>FILENAME</name>
            </field>
            <field>
              <id>OBJECT_ID</id>
              <enabled>Y</enabled>
              <name>OBJECT_ID</name>
            </field>
            <field>
              <id>OBJECT_REVISION</id>
              <enabled>Y</enabled>
              <name>OBJECT_REVISION</name>
            </field>
            <field>
              <id>PARENT_CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>PARENT_CHANNEL_ID</name>
            </field>
            <field>
              <id>ROOT_CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>ROOT_CHANNEL_ID</name>
            </field>
          </channel-log-table>
          <step-log-table>
            <connection/>
            <schema/>
            <table/>
            <timeout_days/>
            <field>
              <id>ID_BATCH</id>
              <enabled>Y</enabled>
              <name>ID_BATCH</name>
            </field>
            <field>
              <id>CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>CHANNEL_ID</name>
            </field>
            <field>
              <id>LOG_DATE</id>
              <enabled>Y</enabled>
              <name>LOG_DATE</name>
            </field>
            <field>
              <id>TRANSNAME</id>
              <enabled>Y</enabled>
              <name>TRANSNAME</name>
            </field>
            <field>
              <id>STEPNAME</id>
              <enabled>Y</enabled>
              <name>STEPNAME</name>
            </field>
            <field>
              <id>STEP_COPY</id>
              <enabled>Y</enabled>
              <name>STEP_COPY</name>
            </field>
            <field>
              <id>LINES_READ</id>
              <enabled>Y</enabled>
              <name>LINES_READ</name>
            </field>
            <field>
              <id>LINES_WRITTEN</id>
              <enabled>Y</enabled>
              <name>LINES_WRITTEN</name>
            </field>
            <field>
              <id>LINES_UPDATED</id>
              <enabled>Y</enabled>
              <name>LINES_UPDATED</name>
            </field>
            <field>
              <id>LINES_INPUT</id>
              <enabled>Y</enabled>
              <name>LINES_INPUT</name>
            </field>
            <field>
              <id>LINES_OUTPUT</id>
              <enabled>Y</enabled>
              <name>LINES_OUTPUT</name>
            </field>
            <field>
              <id>LINES_REJECTED</id>
              <enabled>Y</enabled>
              <name>LINES_REJECTED</name>
            </field>
            <field>
              <id>ERRORS</id>
              <enabled>Y</enabled>
              <name>ERRORS</name>
            </field>
            <field>
              <id>LOG_FIELD</id>
              <enabled>N</enabled>
              <name>LOG_FIELD</name>
            </field>
          </step-log-table>
          <metrics-log-table>
            <connection/>
            <schema/>
            <table/>
            <timeout_days/>
            <field>
              <id>ID_BATCH</id>
              <enabled>Y</enabled>
              <name>ID_BATCH</name>
            </field>
            <field>
              <id>CHANNEL_ID</id>
              <enabled>Y</enabled>
              <name>CHANNEL_ID</name>
            </field>
            <field>
              <id>LOG_DATE</id>
              <enabled>Y</enabled>
              <name>LOG_DATE</name>
            </field>
            <field>
              <id>METRICS_DATE</id>
              <enabled>Y</enabled>
              <name>METRICS_DATE</name>
            </field>
            <field>
              <id>METRICS_CODE</id>
              <enabled>Y</enabled>
              <name>METRICS_CODE</name>
            </field>
            <field>
              <id>METRICS_DESCRIPTION</id>
              <enabled>Y</enabled>
              <name>METRICS_DESCRIPTION</name>
            </field>
            <field>
              <id>METRICS_SUBJECT</id>
              <enabled>Y</enabled>
              <name>METRICS_SUBJECT</name>
            </field>
            <field>
              <id>METRICS_TYPE</id>
              <enabled>Y</enabled>
              <name>METRICS_TYPE</name>
            </field>
            <field>
              <id>METRICS_VALUE</id>
              <enabled>Y</enabled>
              <name>METRICS_VALUE</name>
            </field>
          </metrics-log-table>
        </log>
        <maxdate>
          <connection/>
          <table/>
          <field/>
          <offset>0.0</offset>
          <maxdiff>0.0</maxdiff>
        </maxdate>
        <size_rowset>10000</size_rowset>
        <sleep_time_empty>50</sleep_time_empty>
        <sleep_time_full>50</sleep_time_full>
        <unique_connections>N</unique_connections>
        <feedback_shown>Y</feedback_shown>
        <feedback_size>50000</feedback_size>
        <using_thread_priorities>Y</using_thread_priorities>
        <shared_objects_file/>
        <capture_step_performance>N</capture_step_performance>
        <step_performance_capturing_delay>1000</step_performance_capturing_delay>
        <step_performance_capturing_size_limit>100</step_performance_capturing_size_limit>
        <dependencies>
        </dependencies>
        <partitionschemas>
        </partitionschemas>
        <slaveservers>
        </slaveservers>
        <clusterschemas>
        </clusterschemas>
        <created_user>-</created_user>
        <created_date>2019/07/11 16:30:05.348</created_date>
        <modified_user>-</modified_user>
        <modified_date>2019/07/11 16:30:05.348</modified_date>
        <key_for_session_key>H4sIAAAAAAAAAAMAAAAAAAAAAAA=</key_for_session_key>
        <is_key_private>N</is_key_private>
      </info>
      <notepads>
      </notepads>
      <connection>
        <name>输出数据库</name>
        <server>${dbIp}</server>
        <type>REDSHIFT</type>
        <access>Native</access>
        <database>${dbName}</database>
        <port>${dbPort}</port>
        <username>${dbUsername}</username>
        <password>${dbPassword}</password>
        <servername/>
        <data_tablespace/>
        <index_tablespace/>
        <attributes>
          <attribute>
            <code>EXTRA_OPTION_REDSHIFT.tcpKeepAlive</code>
            <attribute>true</attribute>
          </attribute>
          <attribute>
            <code>FORCE_IDENTIFIERS_TO_LOWERCASE</code>
            <attribute>N</attribute>
          </attribute>
          <attribute>
            <code>FORCE_IDENTIFIERS_TO_UPPERCASE</code>
            <attribute>N</attribute>
          </attribute>
          <attribute>
            <code>IS_CLUSTERED</code>
            <attribute>N</attribute>
          </attribute>
          <attribute>
            <code>PORT_NUMBER</code>
            <attribute>5439</attribute>
          </attribute>
          <attribute>
            <code>PRESERVE_RESERVED_WORD_CASE</code>
            <attribute>Y</attribute>
          </attribute>
          <attribute>
            <code>QUOTE_ALL_FIELDS</code>
            <attribute>N</attribute>
          </attribute>
          <attribute>
            <code>SQL_CONNECT</code>
            <attribute>set names utf8;</attribute>
          </attribute>
          <attribute>
            <code>SUPPORTS_BOOLEAN_DATA_TYPE</code>
            <attribute>Y</attribute>
          </attribute>
          <attribute>
            <code>SUPPORTS_TIMESTAMP_DATA_TYPE</code>
            <attribute>Y</attribute>
          </attribute>
          <attribute>
            <code>USE_POOLING</code>
            <attribute>N</attribute>
          </attribute>
        </attributes>
      </connection>
      <order>
      </order>
      <step>
        <name>执行SQL脚本</name>
        <type>ExecSQL</type>
        <description/>
        <distribute>Y</distribute>
        <custom_distribution/>
        <copies>1</copies>
        <partitioning>
          <method>none</method>
          <schema_name/>
        </partitioning>
        <connection>输出数据库</connection>
        <execute_each_row>N</execute_each_row>
        <single_statement>N</single_statement>
        <replace_variables>N</replace_variables>
        <quoteString>N</quoteString>
        <sql>INSERT INTO `t_user` ( `name`, `age`) VALUES ('王', '99');</sql>
        <set_params>N</set_params>
        <insert_field/>
        <update_field/>
        <delete_field/>
        <read_field/>
        <arguments>
        </arguments>
        <attributes/>
        <cluster_schema/>
        <remotesteps>
          <input>
          </input>
          <output>
          </output>
        </remotesteps>
        <GUI>
          <xloc>64</xloc>
          <yloc>32</yloc>
          <draw>Y</draw>
        </GUI>
      </step>
      <step_error_handling>
      </step_error_handling>
      <slave-step-copy-partition-distribution>
      </slave-step-copy-partition-distribution>
      <slave_transformation>N</slave_transformation>
      <attributes/>
    </transformation>

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center]: /images/20211114/8856086870b342b99f337d9bd7798776.png
[20191014170427325.png_pic_center]: /images/20211114/794bf5137ef2405189c79731cb64ac9e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20211114/bf0197168ac64ce9887b353fcd685039.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20211114/e974472508fd4f4a807db69d2dc7c635.png
[20191014173220445.png_pic_center]: /images/20211114/71a576fad7e84b8ab3c3546437ddc347.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20211114/798e1b277b8a4cb1998447725550c61d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RpYW5kaXh1YW53dWxpYW5n_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20211114/d690db7163f64076a576e0a24f0a5d7f.png
[20191017144537463.png_pic_center]: /images/20211114/9cfac2c5b3c04dea85ef7471743d95d3.png