SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()

.setHost("localhost")
.setDatabase("mydatabase")
.setUsername("myusername")
.setPassword("mypassword")
.setQuery("SELECT message FROM mytable WHERE timestamp > '2021-01-01'")
.setDeserializer(new StringDeserializationSchema())
.build();

这是一个使用 Apache Flink 连接 PostgreSQL 数据库并读取数据的示例代码。该代码使用了 PostgreSQLSource 构建器，并设置了连接信息、查询语句和反序列化器。在这个示例中，我们使用 StringDeserializationSchema 作为反序列化器，因为我们希望读取的数据是字符串类型的。