本文主要介绍Java中,使用JDBC将CSV文件中数据,拷贝复制到Postgres中具有自定义类型的表(table)的方法代码。

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 class 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.class).getCopyAPI();
connection.setAutoCommit(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}]}]