datax扩展vertica插件

╰+哭是因爲堅強的太久メ 2023-09-28 13:41 82阅读 0赞

1.idea里面新建module

![6257fd0e448b4176a3ddc5ee74e29519.png][]

2.新建verticareader模块

![7c68407305744cbcbfea39c3287ffc94.png][]

3.复制mysqlreader下的pom.xml到本项目的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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>com.alibaba.datax</groupId>
            <artifactId>datax-all</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </parent>
        <artifactId>verticareader</artifactId>
        <name>verticareader</name>
        <packaging>jar</packaging>
    
        <dependencies>
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>datax-common</artifactId>
                <version>${datax-project-version}</version>
                <exclusions>
                    <exclusion>
                        <artifactId>slf4j-log4j12</artifactId>
                        <groupId>org.slf4j</groupId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba.datax</groupId>
                <artifactId>plugin-rdbms-util</artifactId>
                <version>${datax-project-version}</version>
            </dependency>
            <dependency>
                <groupId>com.vertica</groupId>
                <artifactId>vertica-jdbc</artifactId>
                <version>9.3.1-0</version>
            </dependency>
    
    
        </dependencies>
    
        <build>
            <plugins>
                
                <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>${jdk-version}</source>
                        <target>${jdk-version}</target>
                        <encoding>${project-sourceEncoding}</encoding>
                    </configuration>
                </plugin>
                
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <descriptors>
                            <descriptor>src/main/assembly/package.xml</descriptor>
                        </descriptors>
                        <finalName>datax</finalName>
                    </configuration>
                    <executions>
                        <execution>
                            <id>dwzip</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    </project>

4.新建assembly目录

![053d86aead2a47c58fd88476d63bced4.png][]

5.目录下新建package.xml,将此文件粘贴进去

<assembly
            xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
        <id></id>
        <formats>
            <format>dir</format>
        </formats>
        <includeBaseDirectory>false</includeBaseDirectory>
        <fileSets>
            <fileSet>
                <directory>src/main/resources</directory>
                <includes>
                    <include>plugin.json</include>
                    <include>plugin_job_template.json</include>
                </includes>
                <outputDirectory>plugin/reader/verticareader</outputDirectory>
            </fileSet>
            <fileSet>
                <directory>target/</directory>
                <includes>
                    <include>verticareader-0.0.1-SNAPSHOT.jar</include>
                </includes>
                <outputDirectory>plugin/reader/verticareader</outputDirectory>
            </fileSet>
        </fileSets>
    
        <dependencySets>
            <dependencySet>
                <useProjectArtifact>false</useProjectArtifact>
                <outputDirectory>plugin/reader/verticareader/libs</outputDirectory>
                <scope>runtime</scope>
            </dependencySet>
        </dependencySets>
    </assembly>

新建plugin.json

{
      "name": "verticareader",
      "class": "com.xxxx.VerticaReader",
      "description": "vertica reader",
      "developer": "cx"
    }

新建plugin\_job\_template.json

{
        "name": "verticareader",
        "parameter": {
            "username": "",
            "password": "",
            "column": [],
            "connection": [
                {
                "jdbcUrl": [],
                "table": []
                }
            ],
            "where": ""
        }
    }

6.src目录新建java文件，包名可以随便取

import com.alibaba.datax.common.element.StringColumn;
    import com.alibaba.datax.common.exception.DataXException;
    import com.alibaba.datax.common.plugin.RecordSender;
    import com.alibaba.datax.common.spi.Reader;
    import com.alibaba.datax.common.util.Configuration;
    import com.alibaba.datax.common.element.Record;
    import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
    import com.alibaba.datax.plugin.rdbms.reader.Constant;
    import com.alibaba.datax.plugin.rdbms.reader.Key;
    import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
    import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.sql.Statement;
    import java.util.ArrayList;
    import java.util.List;
    
    public class VerticaReader extends Reader {
    
        private static final DataBaseType DATABASE_TYPE = DataBaseType.Vertica;
    
        public static class Job extends Reader.Job {
    
            private static final Logger LOG = LoggerFactory
                    .getLogger(Job.class);
            private Configuration originalConfig = null;
            private CommonRdbmsReader.Job commonRdbmsReaderJob;
    
            @Override
            public void init() {
                this.originalConfig = super.getPluginJobConf();
                Integer userConfigedFetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE);
                if (userConfigedFetchSize != null) {
                    LOG.warn("对 mysqlreader 不需要配置 fetchSize, mysqlreader 将会忽略这项配置. 如果您不想再看到此警告,请去除fetchSize 配置.");
                }
                this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);
    
                this.commonRdbmsReaderJob = new CommonRdbmsReader.Job(DATABASE_TYPE);
    //            this.commonRdbmsReaderJob.init(this.originalConfig);
            }
    
            @Override
            public void prepare() {
                LOG.info("vertica prepare....");
                init();
    //            this.commonRdbmsReaderJob.preCheck(this.originalConfig,DATABASE_TYPE);
            }
    
            @Override
            public List<Configuration> split(int adviceNumber) {
                LOG.info("vertica split...."+adviceNumber);
                List<Configuration> configurations = new ArrayList<>();
                Integer partitions = 1;
                for (int i = 0; i < partitions; i++) {
                    configurations.add(this.originalConfig.clone());
                }
                return configurations;
            }
    
            @Override
            public void post() {
                LOG.info("vertica post....");
                this.commonRdbmsReaderJob.post(this.originalConfig);
            }
    
            @Override
            public void destroy() {
                LOG.info("vertica destroy....");
                this.commonRdbmsReaderJob.destroy(this.originalConfig);
            }
    
        }
    
        public static class Task extends Reader.Task {
            private static final Logger LOG = LoggerFactory
                    .getLogger(Task.class);
            private Configuration readerSliceConfig;
            private CommonRdbmsReader.Task commonRdbmsReaderTask;
    
            static final String JDBC_DRIVER="com.vertica.jdbc.Driver";
    
            static  String  DB_URL="jdbc:vertica://xxxxxx:5433/test1";
    
            static  String USER = "xxxx";
    
            static  String PASSWORD = "xxxxx";
    
            static String sql = "";
    
            Connection conn = null;
            Statement stmt = null;
    
            public void doCreateParams(Configuration readerSliceConfig) {
                this.USER = readerSliceConfig.getString(Key.USERNAME);
                this.PASSWORD = readerSliceConfig.getString(Key.PASSWORD);
    
                List<Object> conns = readerSliceConfig.getList(Constant.CONN_MARK, Object.class);
    
                List<Configuration> splittedConfigs = new ArrayList<Configuration>();
                Configuration connConf = Configuration.from(conns.get(0).toString());
                this.DB_URL = connConf.getList(Key.JDBC_URL).get(0).toString();
                this.sql = connConf.getList(Key.QUERY_SQL).get(0).toString();
    //            for (int i = 0, len = conns.size(); i < len; i++) {
    //                Configuration sliceConfig = originalSliceConfig.clone();
    //
    //                Configuration connConf = Configuration.from(conns.get(i).toString());
    //                String jdbcUrl = connConf.getString(Key.JDBC_URL);
    //            }
    
    
                LOG.info("this is ob1_0 jdbc url. user=" + this.USER + " :url=" + this.DB_URL);
    
            }
    
            @Override
            public void init() {
                this.readerSliceConfig = super.getPluginJobConf();
                this.commonRdbmsReaderTask = new CommonRdbmsReader.Task(DATABASE_TYPE,super.getTaskGroupId(), super.getTaskId());
                this.doCreateParams(this.readerSliceConfig);
    
                try{
                    //注册 JDBC 驱动
                    Class.forName(JDBC_DRIVER);
                    //打开链接
                    System.out.println("Connect DB..."+this.USER+","+this.PASSWORD+","+this.DB_URL);
                    conn= DriverManager.getConnection(DB_URL,USER,PASSWORD);
    
                    //执行查询
                    System.out.println("实例化Statement对象...");
                    stmt = conn.createStatement();
    
                }
                catch(Exception e){
                    e.printStackTrace();
                }
    
            }
    
            @Override
            public void prepare() {
                LOG.info("vertica Task prepare....");
            }
    
            @Override
            public void startRead(RecordSender recordSender) {
    //            int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);
    //
    //            this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender,
    //                    super.getTaskPluginCollector(), fetchSize);
                try {
                    String table = readerSliceConfig.getString(Key.TABLE);
                    String columns = readerSliceConfig.getString(Key.COLUMN);
                    LOG.info(table+","+columns+"------------------");
    
                    ResultSet rs = stmt.executeQuery(sql);
                    //展开结果集数据库
                    System.out.println("aoeId" + "\t" + "aoeAes" + "\t" + "aoeSm4");
                    while (rs.next()) {
                        System.out.print(rs.getInt(1) + "\t");
                        System.out.print(rs.getString(2) + "\t");
                        System.out.print(rs.getString(3) + "\t");
                        System.out.println();
                        Record record = recordSender.createRecord();
                        record.addColumn(new StringColumn(rs.getString(1)));
                        record.addColumn(new StringColumn(rs.getString(2)));
                        record.addColumn(new StringColumn(rs.getString(3)));
                        recordSender.sendToWriter(record);
                    }
                    rs.close();
    
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
    
            @Override
            public void post() {
                LOG.info("vertica Task post....");
                this.commonRdbmsReaderTask.post(this.readerSliceConfig);
            }
    
            @Override
            public void destroy() {
                try {
                    stmt.close();
                    conn.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
                LOG.info("vertica Task destroy....");
                this.commonRdbmsReaderTask.destroy(this.readerSliceConfig);
            }
        }
    
    }

7.执行命令，生成插件

mvn -U clean package assembly:assembly -Dmaven.test.skip=true

8.将本地datax\\plugin\\reader目录上传到datax服务器目录datax/plugin/reader

![1b1801b010174690993e88d3936a9084.png][]

9.编写.json配置文件

\{  
"job": \{  
"content": \[  
\{  
"reader": \{  
"parameter": \{  
"password": "xxxxxx",  
"connection": \[  
\{  
"jdbcUrl": \[  
"jdbc:vertica://xxxxx:5433/test1"  
\],  
"querySql": \[  
"SELECT id,name as ds,age as server\_id FROM TEST\_01"  
\]  
\}  
\],  
"writeMode": "update",  
"username": "xxxxx"  
\},  
"name": "verticareader"  
\},  
"writer": \{  
"parameter": \{  
"password": "xxxxx",  
"column": \[  
"id",  
"ds",  
"server\_id"  
\],  
"connection": \[  
\{  
"jdbcUrl": "jdbc:mysql://xxxxxx:3306/xxxxx?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true",  
"table": \[  
"xxxxx"  
\]  
\}  
\],  
"writeMode": "update",  
"username": "xxxx"  
\},  
"name": "mysqlwriter",  
"writeMode": "update"  
\}  
\}  
\],  
"setting": \{  
"speed": \{  
"byte": 1048576,  
"channel": 1  
\}  
\}  
\}  
\}

10.执行命令python bin/datax.py job/vertica\_2\_mysql.json

![235743f0b4714e54a057e5375ac12480.png][]

vertica表结构

![46a09ff6ac7c4cfd885f9c4d4f8f91e8.png][]

mysql表结构

![1ad62bcac8044c6aa9be7932b0bd41ad.png][]

搞定：）

[6257fd0e448b4176a3ddc5ee74e29519.png]: https://img-blog.csdnimg.cn/6257fd0e448b4176a3ddc5ee74e29519.png
[7c68407305744cbcbfea39c3287ffc94.png]: https://img-blog.csdnimg.cn/7c68407305744cbcbfea39c3287ffc94.png
[053d86aead2a47c58fd88476d63bced4.png]: https://img-blog.csdnimg.cn/053d86aead2a47c58fd88476d63bced4.png
[1b1801b010174690993e88d3936a9084.png]: https://img-blog.csdnimg.cn/1b1801b010174690993e88d3936a9084.png
[235743f0b4714e54a057e5375ac12480.png]: https://img-blog.csdnimg.cn/235743f0b4714e54a057e5375ac12480.png
[46a09ff6ac7c4cfd885f9c4d4f8f91e8.png]: https://img-blog.csdnimg.cn/46a09ff6ac7c4cfd885f9c4d4f8f91e8.png
[1ad62bcac8044c6aa9be7932b0bd41ad.png]: https://img-blog.csdnimg.cn/1ad62bcac8044c6aa9be7932b0bd41ad.png