Spark/Flink 写入达梦数据库的插入更新实践

Spark、Flink 向达梦数据库同步数据时，经常需要实现一种写入逻辑：目标记录存在就更新，不存在就插入。常用实现有两种：使用 MERGE INTO，或者先执行 UPDATE，未更新到记录时再执行 INSERT。

本文通过一个简单示例介绍这两种写法，以及并行任务落库时需要注意的问题。

一、准备目标表

假设计算结果需要写入客户状态表：

CREATE TABLE APP.CUSTOMER_CURRENT (
    TENANT_ID       VARCHAR2(32) NOT NULL,
    CUSTOMER_ID     VARCHAR2(64) NOT NULL,
    CUSTOMER_NAME   VARCHAR2(200),
    SOURCE_VERSION  BIGINT NOT NULL,
    UPDATED_AT      TIMESTAMP(6) NOT NULL,
    CONSTRAINT PK_CUSTOMER_CURRENT
        PRIMARY KEY (TENANT_ID, CUSTOMER_ID)
);

TENANT_ID 和 CUSTOMER_ID 组成业务唯一键。主键不仅用于查询，也可以在多个 Spark 分区或 Flink Sink 并发写入相同客户时阻止重复数据。

二、使用 MERGE INTO

MERGE INTO 可以在一条 SQL 中完成匹配、更新和插入：

MERGE INTO APP.CUSTOMER_CURRENT t
USING (
    SELECT
        ? AS TENANT_ID,
        ? AS CUSTOMER_ID,
        ? AS CUSTOMER_NAME,
        ? AS SOURCE_VERSION,
        ? AS UPDATED_AT
    FROM DUAL
) s
ON (
    t.TENANT_ID = s.TENANT_ID
    AND t.CUSTOMER_ID = s.CUSTOMER_ID
)
WHEN MATCHED THEN UPDATE SET
    t.CUSTOMER_NAME  = s.CUSTOMER_NAME,
    t.SOURCE_VERSION = s.SOURCE_VERSION,
    t.UPDATED_AT     = s.UPDATED_AT
WHERE s.SOURCE_VERSION > t.SOURCE_VERSION
WHEN NOT MATCHED THEN INSERT (
    TENANT_ID,
    CUSTOMER_ID,
    CUSTOMER_NAME,
    SOURCE_VERSION,
    UPDATED_AT
) VALUES (
    s.TENANT_ID,
    s.CUSTOMER_ID,
    s.CUSTOMER_NAME,
    s.SOURCE_VERSION,
    s.UPDATED_AT
);

执行逻辑如下：

业务键已存在，执行 UPDATE；
业务键不存在，执行 INSERT；
已存在记录的版本号大于或等于传入版本时，不执行更新。

版本条件可以避免任务重跑或消息乱序时，旧数据覆盖新数据。需要注意，出现在 ON 条件中的业务键不能在更新分支中修改。

在 Flink JDBC 自定义方言中，可以在 getUpsertStatement 中生成这条 SQL：唯一键用于拼接 ON 条件，其他字段用于拼接 UPDATE SET。项目中的达梦方言采用的就是这种实现方式。

三、先 UPDATE 再 INSERT

另一种方式是先更新目标记录：

UPDATE APP.CUSTOMER_CURRENT
SET CUSTOMER_NAME  = ?,
    SOURCE_VERSION = ?,
    UPDATED_AT     = ?
WHERE TENANT_ID = ?
  AND CUSTOMER_ID = ?;

应用读取 JDBC 返回的影响行数。如果为 0，再执行插入：

INSERT INTO APP.CUSTOMER_CURRENT (
    TENANT_ID,
    CUSTOMER_ID,
    CUSTOMER_NAME,
    SOURCE_VERSION,
    UPDATED_AT
) VALUES (?, ?, ?, ?, ?);

两条 SQL 应放在同一个事务中：

connection.setAutoCommit(false);

try {
    int updated = executeUpdate(connection, data);
    if (updated == 0) {
        executeInsert(connection, data);
    }
    connection.commit();
} catch (SQLException e) {
    connection.rollback();
    throw e;
}

并行任务写入同一个新业务键时，两个任务的 UPDATE 都可能返回 0，随后同时执行 INSERT。主键会阻止重复数据，其中一个任务可能收到唯一约束异常。写入端可以回滚当前事务，再在新事务中重新执行完整的插入更新逻辑。

如果 UPDATE 还带有版本条件，影响行数为 0 也可能表示数据库中的版本已经更新，而不一定表示记录不存在。这类场景使用带版本条件的 MERGE 会更直接。

四、Spark/Flink 落库注意事项

1. 按业务键组织数据

尽量让相同业务键进入同一分区或同一个 Sink Subtask，并在写入前按业务键和版本去重。这样可以减少不同连接同时更新相同记录的情况。

2. 保留数据库唯一约束

Spark/Flink 中声明的逻辑主键不能替代达梦表上的物理主键或唯一约束。数据库约束用于防止并发写入产生重复数据。

3. 考虑任务重试

Flink Checkpoint 恢复、Spark Stage 重试以及批次重跑都可能重复提交数据。写入 SQL 需要具备幂等性，可以使用“业务唯一键 + 来源版本号”控制最终结果。

4. 使用批处理和参数绑定

写入时应使用 PreparedStatement 和参数绑定。数据量较大时可以先批量写入暂存表，按业务键去重后，再通过一条 MERGE 合并到目标表。

五、小结

Spark/Flink 写入达梦数据库时，可以使用以下两种插入更新方式：

MERGE INTO：在一条 SQL 中完成更新和插入，也方便增加版本判断；
先 UPDATE 再 INSERT：实现直观，但要使用事务，并处理并发插入产生的唯一约束异常。

无论采用哪种方式，目标表都应建立与业务键一致的主键或唯一约束。对于可能重跑、重复投递或乱序到达的数据，可以同时保存来源版本号，避免旧数据覆盖新数据。