如何基于Flink Datastream实现PostgreSQL的存在性校验与增量更新

核心思路是利用PostgreSQL原生的INSERT ... ON CONFLICT ... DO UPDATE语法实现UPSERT逻辑，同时通过列值对比仅更新发生变化的字段，避免无意义的写操作。

1. 前提条件 #

• 目标PostgreSQL表必须有主键或唯一约束（用于判断记录是否已存在），假设表user的主键为id，表结构如下：

  CREATE TABLE user (
      id INT PRIMARY KEY,
      name VARCHAR(50),
      age INT,
      email VARCHAR(100)
  );
  

2. 编写UPSERT的DML语句 #

针对需求，构造的SQL需要满足：

• 插入新记录（当id不存在时）
• 若id已存在，仅更新与现有值不同的列，且仅当至少有一列变化时才执行更新

SQL语句示例：

INSERT INTO user(id, name, age, email)
VALUES (?, ?, ?, ?)
ON CONFLICT (id) DO UPDATE
SET 
    name = CASE WHEN user.name IS DISTINCT FROM EXCLUDED.name THEN EXCLUDED.name ELSE user.name END,
    age = CASE WHEN user.age IS DISTINCT FROM EXCLUDED.age THEN EXCLUDED.age ELSE user.age END,
    email = CASE WHEN user.email IS DISTINCT FROM EXCLUDED.email THEN EXCLUDED.email ELSE user.email END
WHERE (user.name, user.age, user.email) IS DISTINCT FROM (EXCLUDED.name, EXCLUDED.age, EXCLUDED.email)

• EXCLUDED关键字指代待插入的新记录行
• IS DISTINCT FROM用于处理NULL值的比较（普通!=无法正确判断NULL）
• 末尾的WHERE子句确保只有当至少一列值发生变化时才执行更新，避免无效写操作

3. Flink Datastream代码实现 #

将上述SQL整合到Flink JDBC Sink中，完整代码示例如下：

定义实体类 #

public class User {
    private Integer id;
    private String name;
    private Integer age;
    private String email;

    // Getters, Setters, 全参/无参构造方法
    public Integer getId() { return id; }
    public void setId(Integer id) { this.id = id; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
}

主流程代码 #

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.jdbc.JdbcConnectionOptions;
import org.apache.flink.streaming.connectors.jdbc.JdbcExecutionOptions;
import org.apache.flink.streaming.connectors.jdbc.JdbcSink;
import com.alibaba.fastjson.JSON;
import java.util.Properties;

public class KafkaToPostgresUpsert {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka配置
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "flink-kafka-group");
        kafkaProps.setProperty("auto.offset.reset", "latest");

        // 从Kafka读取JSON格式的用户数据，转换为User对象
        DataStream<User> userStream = env
                .addSource(new FlinkKafkaConsumer<>("user_topic", new SimpleStringSchema(), kafkaProps))
                .map(jsonStr -> JSON.parseObject(jsonStr, User.class));

        // 配置JDBC Sink
        JdbcSink<User> jdbcSink = JdbcSink.sink(
                // 传入之前定义的UPSERT SQL
                "INSERT INTO user(id, name, age, email) VALUES (?, ?, ?, ?) ON CONFLICT (id) DO UPDATE SET name = CASE WHEN user.name IS DISTINCT FROM EXCLUDED.name THEN EXCLUDED.name ELSE user.name END, age = CASE WHEN user.age IS DISTINCT FROM EXCLUDED.age THEN EXCLUDED.age ELSE user.age END, email = CASE WHEN user.email IS DISTINCT FROM EXCLUDED.email THEN EXCLUDED.email ELSE user.email END WHERE (user.name, user.age, user.email) IS DISTINCT FROM (EXCLUDED.name, EXCLUDED.age, EXCLUDED.email)",
                // 参数映射：将User对象的属性绑定到SQL的占位符
                (ps, user) -> {
                    ps.setInt(1, user.getId());
                    ps.setString(2, user.getName());
                    ps.setInt(3, user.getAge());
                    ps.setString(4, user.getEmail());
                },
                // 批量执行配置，提升写入性能
                JdbcExecutionOptions.builder()
                        .withBatchSize(100)
                        .withBatchIntervalMs(500)
                        .withMaxRetries(3)
                        .build(),
                // PostgreSQL连接配置
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:postgresql://localhost:5432/your_db")
                        .withDriverName("org.postgresql.Driver")
                        .withUsername("your_user")
                        .withPassword("your_password")
                        .build()
        );

        // 将数据流写入PostgreSQL
        userStream.addSink(jdbcSink);

        env.execute("Kafka to PostgreSQL Upsert Job");
    }
}

4. 关键注意事项 #

• NULL值处理：必须使用IS DISTINCT FROM而不是!=，否则NULL值的变化无法被正确识别
• 性能优化：根据业务场景调整批量写入的batchSize和batchIntervalMs，平衡延迟和吞吐量
• 约束检查：确保目标表的主键/唯一约束正确配置，否则ON CONFLICT子句会失效
• 依赖引入：需要在项目中添加Flink JDBC和PostgreSQL驱动依赖，比如Maven依赖：

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

内容的提问来源于stack exchange，提问作者Abhinav