Как написать Postgres Sink с Apache Flink 2.0 ⇐ JAVA

[Anonymous]

Я крутил колеса на пару дней. Я думаю, что документация не соответствует API из того, что я вижу - и ссылки на Java DOC, по -видимому, нарушены. https://nightlies.apache.org/flink/flin ... азной[code]

org.apache.flink
flink-connector-jdbc-core
4.0.0-2.0




org.apache.flink
flink-connector-jdbc-postgres
4.0.0-2.0

< /code>
И просто чтобы показать, что я сам пробовал сам здесь - моя попытка: < /p>
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 void addSinkToStream(DataStream stream, JdbcStatementBuilder 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 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 stream) {
addSinkToStream(stream, (preparedStatement, jsonObject) -> {
// Customize this based on your table schema and JSON structure
preparedStatement.setObject(1, jsonObject.toString());
});
}
}
[/code]
Любая помощь будет очень оценена.

Подробнее здесь: https://stackoverflow.com/questions/796 ... -flink-2-0