Flink Table左外连接报错，内连接可正常转换为DataStream

问题原因 #

左外连接会生成包含**更新（UPDATE）、删除（DELETE）类型的变更日志（changelog），而内连接仅输出插入（INSERT）**类型的数据。默认的DataStream sink是Append模式，只能处理INSERT类型数据，因此无法消费左外连接产生的changelog，触发异常。

解决办法 #

1. 改用变更日志流转换API #

放弃直接调用toDataStream()，改用toChangelogStream或toRetractStream处理带变更的流：

// 方式1：获取包含变更类型的数据流
DataStream<Row> changelogStream = tableEnv.toChangelogStream(table);

// 方式2：获取标记变更类型的二元组流（true=插入，false=删除/更新旧数据）
DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);

之后可根据业务逻辑处理这些变更（比如更新外部存储的对应记录）。

2. 使用支持Upsert模式的Sink #

如果需要将结果写入外部存储，选择支持Upsert的Table Sink（如Kafka、HBase等），配置为Upsert模式处理更新/删除操作。以Kafka为例：

tableEnv.executeSql("""
    CREATE TABLE result_sink (
        customermessage STRING,
        contactmessage STRING,
        PRIMARY KEY (customermessage) NOT ENFORCED -- 指定主键用于Upsert匹配
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'result_topic',
        'properties.bootstrap.servers' = 'localhost:9092',
        'key.format' = 'json',
        'value.format' = 'json',
        'sink.mode' = 'upsert' -- 开启Upsert模式
    )
""");

table.executeInsert("result_sink");

3. 改用窗口左外连接（适用于带时间属性的场景） #

如果数据带有时间属性，可使用窗口左外连接，这种连接仅在窗口结束时输出一次快照结果，生成Append流：

Table windowLeftJoinTable = customerTable
    .leftOuterJoin(contactTable, 
        $("cust_custcode").isEqual($("contact_custcode"))
        .and($("customer_time").between($("contact_window_start"), $("contact_window_end")))
    )
    .select($("customermessage"), $("contactmessage"));

适合不需要实时更新连接结果，仅需窗口内数据快照的业务场景。

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