Java使用JDBC将CSV文件中数据复制(copy)到Postgres

1､在Postgresql中创建自定义类型和表

CREATE TYPE address AS (ip inet, port int);

创建有自定义类型的表

CREATE TABLE my_table (
  addresses  address[] NULL
)

2､COPY的DML和CSV文件格式内容

不能解析CSV时，将逗号解释为字段定界符。让CSV解析器将整个行视为一个字符串，一个字段，可以通过将其用单引号引起来或者让PG字段解析器考虑将每个数组项目类型实例括在双引号中。

1) 使用单引号的DML

COPY my_table (addresses) FROM STDIN WITH CSV QUOTE ''''

2) CSV文件

'{"(10.0.0.1,1)","(10.0.0.2,2)"}'
'{"(10.10.10.1,80)","(10.10.10.2,443)"}'
'{"(10.10.10.3,8080)","(10.10.10.4,4040)"}'

或者

1) 使用双引号的DML，转义双引号

COPY my_table (addresses) FROM STDIN WITH CSV

2) CSV文件，转义双引号

"{""(10.0.0.1,1)"",""(10.0.0.2,2)""}"
"{""(10.10.10.1,80)"",""(10.10.10.2,443)""}"
"{""(10.10.10.3,8080)"",""(10.10.10.4,4040)""}"

3､实现代码

完整的JUnit测试类

package io.mikael.poc;
import com.google.common.io.CharStreams;
import org.junit.*;
import org.postgresql.PGConnection;
import org.postgresql.copy.CopyManager;
import org.testcontainers.containers.PostgreSQLContainer;
import java.io.*;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import static java.nio.charset.StandardCharsets.UTF_8;
public >CopyTest {
    private Reader reader;
    private Connection connection;
    private CopyManager copyManager;
    private static final String CREATE_TYPE = "CREATE TYPE address AS (ip inet, port int)";
    private static final String CREATE_TABLE = "CREATE TABLE my_table (addresses  address[] NULL)";
    private String loadCsvFromFile(final String fileName) throws IOException {
        try (InputStream is = getClass().getResourceAsStream(fileName)) {
            return CharStreams.toString(new InputStreamReader(is, UTF_8));
        }
    }
    @ClassRule
    public static PostgreSQLContainer db = new PostgreSQLContainer("postgres:10-alpine");
    @BeforeClass
    public static void beforeClass() throws Exception {
        Class.forName("org.postgresql.Driver");
    }
    @Before
    public void before() throws Exception {
        String input = loadCsvFromFile("/data_01.csv");
        reader = new StringReader(input);
        connection = DriverManager.getConnection(db.getJdbcUrl(), db.getUsername(), db.getPassword());
        copyManager = connection.unwrap(PGConnection.>false);
        connection.beginRequest();
        connection.prepareCall(CREATE_TYPE).execute();
        connection.prepareCall(CREATE_TABLE).execute();
    }
    @After
    public void after() throws Exception {
        connection.rollback();
    }
    @Test
    public void copyTest01() throws Exception {
        copyManager.copyIn("COPY my_table (addresses) FROM STDIN WITH CSV QUOTE ''''", reader);
        final StringWriter writer = new StringWriter();
        copyManager.copyOut("COPY my_table TO STDOUT WITH CSV", writer);
        System.out.printf("roundtrip:%n%s%n", writer.toString());
        final ResultSet rs = connection.prepareStatement(
                "SELECT array_to_json(array_agg(t)) FROM (SELECT addresses FROM my_table) t")
                .executeQuery();
        rs.next();
        System.out.printf("json:%n%s%n", rs.getString(1));
    }
}

测试输出：

roundtrip:
"{""(10.0.0.1,1)"",""(10.0.0.2,2)""}"
"{""(10.10.10.1,80)"",""(10.10.10.2,443)""}"
"{""(10.10.10.3,8080)"",""(10.10.10.4,4040)""}"
json:
[{"addresses":[{"ip":"10.0.0.1","port":1},{"ip":"10.0.0.2","port":2}]},{"addresses":[{"ip":"10.10.10.1","port":80},{"ip":"10.10.10.2","port":443}]},{"addresses":[{"ip":"10.10.10.3","port":8080},{"ip":"10.10.10.4","port":4040}]}]