flink 复postgresql数据库数据

要使用Apache Flink复制PostgreSQL数据库的数据，你可以使用Flink的CDC（Change Data Capture）功能来监听数据库的变更，并将这些变更实时同步到Flink程序中。以下是一个简单的例子，展示如何使用Flink的Debezium连接器来复制PostgreSQL的数据。

首先，确保你的PostgreSQL支持逻辑复制，并且Flink的Debezium连接器已经包含在你的Flink工程中。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-debezium</artifactId>
    <version>${flink.version}</version>
</dependency>

然后，你可以使用以下代码片段来创建Flink程序，该程序监听PostgreSQL的变更并输出到控制台：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.DataTypes;
 
public class FlinkCDCExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 
        String dbeventsTableDDL = "" +
                "CREATE TABLE dbevents (" +
                "   id INT," +
                "   data ROW<id INT, name STRING>," +
                "   op STRING," +
                "   ts TIMESTAMP(3)," +
                "   source ROW<version STRING, connector STRING, name STRING, ts MS TIMESTAMP>" +
                ") WITH (" +
                "   'connector' = 'debezium'," +
                "   'format' = 'json'," +
                "   'debezium.io.bootstrappoint.offset' = 'earliest'," +
                "   'debezium.io.bootstrappoint.topic' = 'your_bootstrappoint_topic'," +
                "   'database.name' = 'your_db_name'," +
                "   'database.hostname' = 'your_db_host'," +
                "   'database.port' = 'your_db_port'," +
                "   'table.name' = 'your_table_name'" +
                ")";
 
        tableEnv.executeSql(dbeventsTableDDL);
 
        // 监听变更并输出
        tableEnv.executeSql("INSERT INTO console SELECT id, data.name FROM dbevents WHERE op = 'c'").execute();
 
        env.execute("Flink CDC Example");
    }
}

在这个例子中，我们创建了一个名为dbevents的表，该表连接到Debezium CDC源。然后我们执行了一个SQL查询，它从dbevents表中选择所有插入（操作为'c'）的行，并将这些行的id和data.name字段插入到控制台。

确保替换dbeventsTableDDL中的连接参数，以反映你的PostgreSQL数据库的实际配置。