如何使用Apache Flink 2.0编写Postgres Sink？文档API不符求助

折腾了好几天，发现Flink 2.0官方JDBC连接器文档和实际API不匹配，且JavaDoc链接已失效。我需要实现一个Sink，将流中的JSON数据保存到Postgres数据库中。

依赖配置 #

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc-core -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc-core</artifactId>
    <version>4.0.0-2.0</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc-postgres -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc-postgres</artifactId>
    <version>4.0.0-2.0</version>
</dependency>

尝试的代码 #

package com.ragdist;

import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.json.JSONObject;

public class PgDocSink {
    private final String jdbcUrl;
    private final String username;
    private final String password;
    private final String tableName;

    public PgDocSink(String jdbcUrl, String username, String password, String tableName) {
        this.jdbcUrl = jdbcUrl;
        this.username = username;
        this.password = password;
        this.tableName = tableName;
    }

    public <T> void addSinkToStream(DataStream<T> stream, JdbcStatementBuilder<T> statementBuilder) {
        // Create execution options
        JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
                .withBatchSize(1000)
                .withBatchIntervalMs(200)
                .withMaxRetries(5)
                .build();

        // Connection options
        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(jdbcUrl)
                .withDriverName("org.postgresql.Driver")
                .withUsername(username)
                .withPassword(password)
                .build();

        JdbcExactlyOnceOptions exactlyOnceOptions = JdbcExactlyOnceOptions.builder()
                .withTransactionPerConnection(true)
                .build();

        // ---------- problem area - don't know what the right api ---------
        // Use the non-deprecated JdbcSink from the recommended package
        Sink<T> sink = JdbcSink.builder().buildExactlyOnce(exactlyOnceOptions, null)

                // .sink(
                //         String.format("INSERT INTO %s VALUES (?)", tableName),
                //         statementBuilder,
                //         executionOptions,
                //         connectionOptions);

        // ---------- -------------------------------------------- ---------

        // Add sink to the stream
        stream.sinkTo(sink);
    }

    public void addJsonSinkToStream(DataStream<JSONObject> stream) {
        addSinkToStream(stream, (preparedStatement, jsonObject) -> {
            // Customize this based on your table schema and JSON structure
            preparedStatement.setObject(1, jsonObject.toString());
        });
    }
}

解决方案 #

针对Flink 4.0.0-2.0版本的JDBC连接器，JdbcSink的构建方式已更新为链式builder模式，以下是正确实现方式：

1. 普通At-Least-Once语义实现 #

如果不需要Exactly-Once语义，直接通过链式配置参数构建Sink：

public <T> void addSinkToStream(DataStream<T> stream, JdbcStatementBuilder<T> statementBuilder) {
    JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
            .withBatchSize(1000)
            .withBatchIntervalMs(200)
            .withMaxRetries(5)
            .build();

    JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl(jdbcUrl)
            .withDriverName("org.postgresql.Driver")
            .withUsername(username)
            .withPassword(password)
            .build();

    String insertSql = String.format("INSERT INTO %s VALUES (?)", tableName);

    Sink<T> sink = JdbcSink.<T>builder()
            .setSql(insertSql)
            .setStatementBuilder(statementBuilder)
            .setExecutionOptions(executionOptions)
            .setConnectionOptions(connectionOptions)
            .build();

    stream.sinkTo(sink);
}

2. Exactly-Once语义实现 #

如果需要Exactly-Once语义，需结合JdbcExactlyOnceOptions，同时确保作业已开启Checkpoint：

public <T> void addSinkToStream(DataStream<T> stream, JdbcStatementBuilder<T> statementBuilder) {
    JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
            .withBatchSize(1000)
            .withBatchIntervalMs(200)
            .withMaxRetries(5)
            .build();

    JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl(jdbcUrl)
            .withDriverName("org.postgresql.Driver")
            .withUsername(username)
            .withPassword(password)
            .build();

    JdbcExactlyOnceOptions exactlyOnceOptions = JdbcExactlyOnceOptions.builder()
            .withTransactionPerConnection(true)
            .build();

    String insertSql = String.format("INSERT INTO %s VALUES (?)", tableName);

    Sink<T> sink = JdbcSink.<T>builder()
            .setSql(insertSql)
            .setStatementBuilder(statementBuilder)
            .setExecutionOptions(executionOptions)
            .setConnectionOptions(connectionOptions)
            .buildExactlyOnce(exactlyOnceOptions)
            .build();

    stream.sinkTo(sink);
}

关键说明 #

• 新版本JdbcSink.builder()采用链式调用配置SQL、StatementBuilder、执行选项、连接选项，最后调用build()生成Sink实例；
• Exactly-Once语义需先调用buildExactlyOnce(exactlyOnceOptions)，再执行build()；
• 必须开启Flink作业的Checkpoint，否则Exactly-Once语义无法生效；
• flink-connector-jdbc-postgres依赖会自动适配Postgres的JDBC特性，无需额外配置。

内容的提问来源于stack exchange，提问作者Edv Beq