Flink JDBC沉入多个模式

Question

我正在使用Flink JDBC接收器将数据推入Postgres表中。数据必须存储在具有相同数据库连接的不同的模式中。

DataStream<Book> stream = env.fromSource(...);

流中的每个书籍记录都有有关必须存储的架构的详细信息。

我试图在准备台词中参数化数据库架构名称，但正如预期的那样，SQL不允许使用。

stream.addSink(JdbcSink.sink(
                    "insert into ?.books (id, title, authors, year) values (?, ?, ?, ?)",
                    (statement, book) -> {
                        statement.setString(1, book.schema)
                        statement.setLong(2, book.id);
                        statement.setString(3, book.title);
                        statement.setString(4, book.authors);
                        statement.setInt(5, book.year);
                    },
                    JdbcExecutionOptions.builder()
                            .withBatchSize(1000)
                            .withBatchIntervalMs(200)
                            .withMaxRetries(5)
                            .build(),
                    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                            .withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
                            .withDriverName("org.postgresql.Driver")
                            .withUsername("someUser")
                            .withPassword("somePassword")
                            .build()
            ));

有解决方法吗？还是我必须在单独的水槽中明确添加每个架构？

Answer 1

假设我们有一系列字符串，其中列出了所有数据库模式（可以通过应用程序属性进行预配置）。

我们可以根据架构名称过滤流，并为每个模式创建一个单独的JDBC接收器。

String[] schemas = {"schema-1", "schema-2", "schema-3"};

for(int i=0; i<schemas.length; i++){
    stream.filter(x -> x.schema.equals(schemas[i])).addSink(JdbcSink.sink(
                "insert into " + schemas[i] ".books (id, title, authors, year) values (?, ?, ?, ?)",
                (statement, book) -> {
                    statement.setLong(1, book.id);
                    statement.setString(2, book.title);
                    statement.setString(3, book.authors);
                    statement.setInt(4, book.year);
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(200)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:postgresql://dbhost:5432/postgresdb")
                        .withDriverName("org.postgresql.Driver")
                        .withUsername("someUser")
                        .withPassword("somePassword")
                        .build()
        ));

}