Apache Hop使用

傷城~ 2024-03-26 21:15 43阅读 0赞

## 1、maven 依赖 ##

### 1.1、本地镜像 ###

<mirror>
        <id>nexus-aliyun</id>
        <mirrorOf>jcenter</mirrorOf>
        <url>https://maven.aliyun.com/repository/central</url>
    </mirror>

### 1.2、pom.xml依赖 ###

<?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 https://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.2.RELEASE</version>
            <relativePath/> 
        </parent>
        <groupId>com.example</groupId>
        <artifactId>demo</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>demo</name>
        <description>Demo project for Spring Boot</description>
        <properties>
            <java.version>1.8</java.version>
            <hop.version>2.3.0</hop.version>
        </properties>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.hop</groupId>
                <artifactId>hop-engine</artifactId>
                <version>${hop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hop</groupId>
                <artifactId>hop-engine-beam</artifactId>
                <version>${hop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hop</groupId>
                <artifactId>hop-core</artifactId>
                <version>${hop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hop</groupId>
                <artifactId>hop-ui</artifactId>
                <version>${hop.version}</version>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.22</version>
            </dependency>
        </dependencies>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

## 2、执行代码 ##

### 2.1、配置文件apllication ###

spring.application.name=hop-demo
    server.port=8888
    
    hop.metadata.folder=D:\\Apache Hop\\apache-hop-client-2.3.0\\hop\\config\\projects\\default\\metadata
    hop.folder=D:\\Apache Hop\\apache-hop-client-2.3.0\\hop\\config\\projects\\default
    hop.plugin=D:\\Apache Hop\\apache-hop-client-2.3.0\\hop\\plugins

### 2.2、启动类 ###

package com.example.demo;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class DemoApplication {
    
        public static void main(String[] args){
            SpringApplication.run(DemoApplication.class, args);
        }
    
    }

### 2.3、hop加载插件、hop初始化 ###

package com.example.demo.service;
    
    
    import org.apache.hop.core.HopEnvironment;
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.core.env.Environment;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    public class HopInit implements ApplicationRunner {
    
        @Resource
        private Environment env;
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
            try{
                // 加载插件
                System.setProperty("HOP_PLUGIN_BASE_FOLDERS",env.getProperty("hop.plugin"));
                // hop初始化
                HopEnvironment.init();
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    }

### 2.4、执行hpl文件测试 ###

package com.example.demo.controller;
    
    import org.apache.commons.lang3.StringUtils;
    import org.apache.hop.core.Result;
    import org.apache.hop.core.encryption.Encr;
    import org.apache.hop.core.encryption.HopTwoWayPasswordEncoder;
    import org.apache.hop.core.encryption.ITwoWayPasswordEncoder;
    import org.apache.hop.core.variables.IVariables;
    import org.apache.hop.core.variables.Variables;
    import org.apache.hop.metadata.api.IHopMetadataProvider;
    import org.apache.hop.metadata.serializer.json.JsonMetadataProvider;
    import org.apache.hop.metadata.serializer.multi.MultiMetadataProvider;
    import org.apache.hop.pipeline.PipelineMeta;
    import org.apache.hop.pipeline.engine.EngineMetrics;
    import org.apache.hop.pipeline.engine.IPipelineEngine;
    import org.apache.hop.pipeline.engine.PipelineEngineFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.io.File;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    import java.util.Map;
    
    @RestController
    @Configuration
    public class PipelineTest {
    
        @Value("${hop.folder}")
        private  String hplFolder;
    
        @Value("${hop.metadata.folder}")
        private  String metadataFolder;
    
        @GetMapping("/test")
        public void test()  {
            try {
                // 参数
                IVariables variables = Variables.getADefaultVariableSpace();
                variables.setVariable("kpi_db", "demo01");
                variables.setVariable("kpi_table", "sex");
                variables.setVariable("mysql_db1", "mysql");
                variables.setVariable("mysql_db2", "demo01");
                variables.setVariable("mysql_host", "127.0.0.1");
                variables.setVariable("mysql_pass", "root1234");
                variables.setVariable("mysql_port", "3306");
                variables.setVariable("mysql_user", "root");
    
                List<IHopMetadataProvider> providers = new ArrayList<>();
                // 管道运行配置本地路径
                // D:\Apache Hop\apache-hop-client-2.3.0\hop\config\projects\default\metadata
                String folder = metadataFolder;
                if (StringUtils.isBlank(folder)) {
                    providers.add(new JsonMetadataProvider());
                } else {
                    ITwoWayPasswordEncoder passwordEncoder = Encr.getEncoder();
                    if (passwordEncoder == null) {
                        passwordEncoder = new HopTwoWayPasswordEncoder();
                    }
                    providers.add(new JsonMetadataProvider(passwordEncoder, folder, variables));
                }
                IHopMetadataProvider metadataProvider = new MultiMetadataProvider(Encr.getEncoder(), providers, variables);
    
                PipelineMeta pipelineMeta = new PipelineMeta(
    //                "D:\\Apache Hop\\apache-hop-client-2.3.0\\hop\\config\\projects\\default\\filename.hpl",
                        hplFolder+ File.separator+"filename.hpl",
                        metadataProvider,
                        variables
                );
                // 管道运行配置本地
                IPipelineEngine pipelineEngine = PipelineEngineFactory.createPipelineEngine(
                        variables,
                        "local",
                        metadataProvider,
                        pipelineMeta
                );
                pipelineEngine.execute();
                pipelineEngine.waitUntilFinished();
    
    
                Result result = pipelineEngine.getResult();
                Date date = pipelineEngine.getExecutionEndDate();
                Map<String, Object> m = pipelineEngine.getExtensionDataMap();
                EngineMetrics engineMetrics = pipelineEngine.getEngineMetrics();
    
                System.out.println(result);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    
    
    
    }

### 2.5、filename.hpl文件 ###

<?xml version="1.0" encoding="UTF-8"?>
    <pipeline>
      <info>
        <name>filename</name>
        <name_sync_with_filename>Y</name_sync_with_filename>
        <description/>
        <extended_description/>
        <pipeline_version/>
        <pipeline_type>Normal</pipeline_type>
        <pipeline_status>0</pipeline_status>
        <parameters>
          <parameter>
            <name>kpi_db</name>
            <default_value>demo01</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>kpi_table</name>
            <default_value>sex</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_db1</name>
            <default_value>mysql</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_db2</name>
            <default_value>demo01</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_host</name>
            <default_value>127.0.0.1</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_pass</name>
            <default_value>root1234</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_port</name>
            <default_value>3306</default_value>
            <description/>
          </parameter>
          <parameter>
            <name>mysql_user</name>
            <default_value>root</default_value>
            <description/>
          </parameter>
        </parameters>
        <capture_transform_performance>N</capture_transform_performance>
        <transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
        <transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
        <created_user>-</created_user>
        <created_date>2023/03/03 11:18:20.201</created_date>
        <modified_user>-</modified_user>
        <modified_date>2023/03/03 11:18:20.201</modified_date>
      </info>
      <notepads>
      </notepads>
      <order>
        <hop>
          <from>表输入</from>
          <to>表输出</to>
          <enabled>Y</enabled>
        </hop>
      </order>
      <transform>
        <name>表输入</name>
        <type>TableInput</type>
        <description/>
        <distribute>Y</distribute>
        <custom_distribution/>
        <copies>1</copies>
        <partitioning>
          <method>none</method>
          <schema_name/>
        </partitioning>
        <connection>demo01</connection>
        <execute_each_row>N</execute_each_row>
        <limit>0</limit>
        <sql>SELECT sex,count(*) result FROM student group by sex</sql>
        <variables_active>N</variables_active>
        <attributes/>
        <GUI>
          <xloc>208</xloc>
          <yloc>256</yloc>
        </GUI>
      </transform>
      <transform>
        <name>表输出</name>
        <type>TableOutput</type>
        <description/>
        <distribute>Y</distribute>
        <custom_distribution/>
        <copies>1</copies>
        <partitioning>
          <method>none</method>
          <schema_name/>
        </partitioning>
        <commit>1000</commit>
        <connection>demo01</connection>
        <fields>
          <field>
            <column_name>sex</column_name>
            <stream_name>sex</stream_name>
          </field>
          <field>
            <column_name>count</column_name>
            <stream_name>result</stream_name>
          </field>
        </fields>
        <ignore_errors>N</ignore_errors>
        <only_when_have_rows>N</only_when_have_rows>
        <partitioning_daily>N</partitioning_daily>
        <partitioning_enabled>N</partitioning_enabled>
        <partitioning_monthly>Y</partitioning_monthly>
        <return_keys>N</return_keys>
        <schema>${kpi_db}</schema>
        <specify_fields>Y</specify_fields>
        <table>${kpi_table}</table>
        <tablename_in_field>N</tablename_in_field>
        <tablename_in_table>Y</tablename_in_table>
        <truncate>N</truncate>
        <use_batch>Y</use_batch>
        <attributes/>
        <GUI>
          <xloc>432</xloc>
          <yloc>256</yloc>
        </GUI>
      </transform>
      <transform_error_handling>
      </transform_error_handling>
      <attributes/>
    </pipeline>