Flink 连接 PostgreSQL 数据库：详细教程和代码示例

要连接 PostgreSQL，您需要添加以下依赖项：

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.2.16</version>
</dependency>

接下来，您可以使用以下代码创建一个 Flink 数据流，将数据写入 PostgreSQL 数据库表：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.connectors.jdbc.JdbcConnectionOptions;
import org.apache.flink.streaming.connectors.jdbc.JdbcSink;
import org.apache.flink.streaming.connectors.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.connectors.jdbc.JdbcSinkBuilder;
import org.apache.flink.streaming.connectors.jdbc.JdbcLookupOptions;
import org.apache.flink.streaming.connectors.jdbc.JdbcInputFormat;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.Jdbc;
import org.apache.flink.types.Row;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

public class FlinkPostgreSQLExample {

    public static void main(String[] args) throws Exception {

        // set up the streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create a table descriptor
        tableEnv.connect(
            new Jdbc()
                .url('jdbc:postgresql://localhost:5432/mydatabase')
                .driver('org.postgresql.Driver')
                .username('myuser')
                .password('mypassword')
                .table('mytable')
                .schema(new Schema()
                    .field('id', 'INT')
                    .field('name', 'STRING')
                    .field('age', 'INT')
                )
        )
        .createTemporaryTable('mytable');

        // create a data stream
        DataStream<Row> stream = env.addSource(new MySourceFunction());

        // convert the data stream to a table
        Table table = tableEnv.fromDataStream(stream);

        // insert the data into the PostgreSQL database
        DataStreamSink<Row> sink = tableEnv.toDataStream(table)
            .addSink(JdbcSink.sink(
                'INSERT INTO mytable (id, name, age) VALUES (?, ?, ?)',
                new JdbcStatementBuilder<Row>() {
                    public void accept(PreparedStatement ps, Row r) throws SQLException {
                        ps.setInt(1, r.getField(0));
                        ps.setString(2, r.getField(1));
                        ps.setInt(3, r.getField(2));
                    }
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                    .withUrl('jdbc:postgresql://localhost:5432/mydatabase')
                    .withDriverName('org.postgresql.Driver')
                    .withUsername('myuser')
                    .withPassword('mypassword')
                    .build()));

        // execute the program
        env.execute();
    }

    // a simple source function that generates some data
    public static class MySourceFunction extends RichParallelSourceFunction<Row> {

        boolean running = true;

        public void run(SourceContext<Row> ctx) throws Exception {
            while (running) {
                ctx.collect(Row.of(1, 'Alice', 25));
                ctx.collect(Row.of(2, 'Bob', 30));
                Thread.sleep(1000);
            }
        }

        public void cancel() {
            running = false;
        }
    }
}

这里使用了 JdbcSink 来将数据写入 PostgreSQL 数据库表。您可以使用 JdbcInputFormat 读取 PostgreSQL 数据库表中的数据。