使用Flink MongoDB SQL Connector跳过Upsert执行普通插入的方案问询

针对你遇到的异常（Join节点产生Update/Delete变更流，但MongoDB Sink不支持处理这类变更），可以通过以下几种方式跳过Upsert语义，强制仅执行普通插入：

1. 在Sink表DDL中显式指定变更日志模式为INSERT_ONLY #

直接在MongoDB Sink表的WITH参数中配置sink.changelog-mode为INSERT_ONLY，强制Connector仅接受并处理INSERT类型的消息，忽略Update/Delete变更：

CREATE TABLE Correlation (
    graph STRING,
    f3 STRING,
    id STRING,
    type STRING,
    startDate TIMESTAMP,
    hascontextEntityId STRING,
    providedBy STRING,
    harvestedDate TIMESTAMP
) WITH (
    'connector' = 'mongodb',
    'uri' = 'mongodb://localhost:27017',
    'database' = 'default_database',
    'collection' = 'Correlation',
    'sink.changelog-mode' = 'INSERT_ONLY' -- 核心配置，强制仅插入
);

这个配置会让连接器放弃Upsert逻辑，将所有收到的消息作为新文档插入MongoDB，不会尝试更新或删除现有文档。

2. 将Join结果转换为Append-only流 #

如果上游Join操作会产生Update/Delete变更，可以通过SQL将结果转换为仅包含INSERT的流。例如用子查询包装Join结果，并指定变更模式：

INSERT INTO Correlation
SELECT * FROM (
    SELECT graph, f3, id, type, startDate, hascontextEntityId, providedBy, harvestedDate
    FROM table_a
    JOIN table_b ON table_a.f3 = table_b.id
) t /*+ OPTIONS('changelog-mode'='INSERT_ONLY') */;

这种方式会告诉Flink planner将Join结果处理为Append-only流，只产生INSERT消息。

3. 切换到DataStream API直接写入 #

如果Table API的配置无法满足需求，可以改用DataStream API的MongoDB Sink，它默认以Append模式工作，只会执行普通插入：

// 假设joinResult是你的Join结果数据流
DataStream<Row> joinResult = ...;

MongoDBSink<Row> mongoSink = MongoDBSink.<Row>builder()
    .setUri("mongodb://localhost:27017")
    .setDatabase("default_database")
    .setCollection("Correlation")
    .setSerializationSchema(new MongoRowSerializationSchema.Builder()
        .setDatabase("default_database")
        .setCollection("Correlation")
        .build())
    .build();

joinResult.addSink(mongoSink);

DataStream的Sink不会处理Upsert逻辑，所有数据都会作为新文档插入MongoDB。

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