Greenplum——copy insert的Java工具类

1、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>

    <groupId>org.example</groupId>
    <artifactId>greenplum-test</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>6</source>
                    <target>6</target>
                </configuration>
            </plugin>
        </plugins>
    </build>


    <dependencies>
        <dependency>
            <groupId>com.pivotal</groupId>
            <artifactId>greenplum-jdbc</artifactId>
            <version>5.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <version>42.2.23</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-dbcp2</artifactId>
            <version>2.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.11</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.83</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.8.0-beta4</version>
        </dependency>
    </dependencies>
</project>

2、GpCopyInDataUtils

package com.greenplum.test.util;

import org.apache.commons.lang3.StringUtils;
import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.io.*;
import java.sql.Connection;
import java.sql.SQLException;

public class GpCopyInDataUtils {

    private static Logger logger = LoggerFactory.getLogger(GpCopyInDataUtils.class);


    /**
     * copy in导出数据
     *
     * @param dataSource 数据源
     * @param querySql   查询sql
     * @param delimiter  列分隔符 例如 ,
     * @param filePath   导出文件路径
     * @param encode     编码 例如 UTF-8
     * @param header     是否需要表头
     * @return
     */
    public static long exportData(DataSource dataSource,
                                  String querySql,
                                  String delimiter,
                                  String filePath,
                                  String encode,
                                  boolean header) {
        Connection con = null;
        OutputStream out = null;
        Writer writer = null;
        try {
            con = dataSource.getConnection();
            CopyManager cm = new CopyManager(con.unwrap(BaseConnection.class));
            StringBuffer sb = new StringBuffer();
            sb.append("copy (");
            sb.append(querySql);
            sb.append(" ) TO STDOUT ");
            sb.append("WITH DELIMITER '");
            sb.append(delimiter);
            sb.append("'");
            if (header) {
                sb.append(" HEADER ");
            }
            String copySql = sb.toString();
            logger.info("exportData data begin ,  sql  is {}", copySql);

            long startTime = System.currentTimeMillis();
            File file = new File(filePath);
            if (!file.getParentFile().exists()) {
                file.getParentFile().mkdirs();
            }
            out = new FileOutputStream(file);
            long handledRowCount = 0;
            if (StringUtils.isNotEmpty(encode)) {
                writer = new OutputStreamWriter(out, encode);
                handledRowCount = cm.copyOut(copySql, writer);
            } else {
                handledRowCount = cm.copyOut(copySql, out);
            }
            long elapsedTime = System.currentTimeMillis() - startTime;

            logger.info("exportData data end, sql  is {}, elapsed time = {}", copySql, elapsedTime);
            return handledRowCount;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return 0L;
        } finally {
            if (out != null) {
                try {
                    out.close();
                } catch (IOException e) {
                    logger.error(e.getMessage(), e);
                }
            }

            if (writer != null) {
                try {
                    writer.close();
                } catch (IOException e) {
                    logger.error(e.getMessage(), e);
                }
            }

            if (con != null) {
                try {
                    con.close();
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

    /**
     * copy in导入数据
     *
     * @param dataSource
     * @param table
     * @param delimiter
     * @param file
     * @return
     */
    public static long importData(DataSource dataSource,
                                  String table,
                                  String delimiter,
                                  String file) {
        Connection con = null;
        InputStream in = null;
        try {
            logger.info("import data begin");
            con = dataSource.getConnection();
            CopyManager cm = new CopyManager(con.unwrap(BaseConnection.class));
            StringBuffer sb = new StringBuffer();
            sb.append("copy ");
            sb.append(table);
            sb.append(" from STDIN  ");
            sb.append("WITH DELIMITER '");
            sb.append(delimiter);
            sb.append("'");
            String copySql = sb.toString();
            logger.info("import data begin,  sql  is {}", copySql);
            long startTime = System.currentTimeMillis();
            in = new FileInputStream(file);
            long handledRowCount = cm.copyIn(copySql, in);
            long elapsedTime = System.currentTimeMillis() - startTime;
            logger.info("import data end,  sql  is {}, elapsed time = {}", copySql, elapsedTime);
            return handledRowCount;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return 0L;
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    logger.error(e.getMessage(), e);
                }
            }

            if (con != null) {
                try {
                    con.close();
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

}
 
 

3、TestMain

package com.greenplum.test;

import com.greenplum.test.util.GpCopyInDataUtils;
import org.apache.commons.dbcp2.BasicDataSource;


public class TestMain {
    public static void main(String[] args) {
        BasicDataSource ds = new BasicDataSource();
        ds.setDriverClassName("org.postgresql.Driver");
        ds.setUrl("jdbc:postgresql://127.0.0.1:54432/pgbenchdb");
        ds.setUsername("gpadmin");
        ds.setPassword("changeme");
        ds.setMaxIdle(10);
        ds.setInitialSize(5);

        // 测试导出表数据到文件
        GpCopyInDataUtils.exportData(ds,
                "select * from tbl",
                ",",
                "/Users/chenweifeng/Downloads/greenplum-test/tmp/tbl.txt",
                "UTF-8",
                true);

        // 测试从文件导入数据到表
        GpCopyInDataUtils.importData(ds,
                "tbl",
                ",",
                "/Users/chenweifeng/Downloads/greenplum-test/tmp/tbl.txt");
    }
}