Flink批模式写入S3表Sink报不支持更新变更错误如何解决？

普通Flink Filesystem连接器默认仅支持Append-only（仅追加） 数据流，你使用的聚合算子会产生包含更新操作的Changelog流，而Filesystem Sink默认未开启变更处理能力，因此无法消费这类流数据触发报错。

• 方案1：适配纯批执行场景，移除输出表主键定义
  你的任务为批处理场景，聚合结果全为最终态的新增数据，不需要主键约束。首先显式将任务执行模式设置为批模式，再删除输出表的主键声明即可：
  首先调整表环境初始化配置：

  import org.apache.flink.configuration.Configuration;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  
  // 初始化时指定批运行模式
  Configuration config = new Configuration();
  config.set("execution.runtime-mode", "BATCH");
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
  StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  

  再修改输出表的DDL，删除PRIMARY KEY配置：

  CREATE TABLE output_table  (
    id STRING,
    created_date DATE,
    count_value BIGINT
  ) WITH (
     'connector' = 'filesystem', 
     'path' = 's3://some-bucket/output-table/',
     'format' = 'json'
   )
  

• 方案2：保留主键配置，开启Filesystem Sink的Upsert能力
  如果你需要保留主键约束，可在输出表的WITH参数中新增sink.upsert-mode = 'true'配置，开启Filesystem连接器的变更处理能力，适配聚合产生的Changelog流：

  CREATE TABLE output_table  (
    id STRING,
    created_date DATE,
    count_value BIGINT,
    PRIMARY KEY (id, created_date) NOT ENFORCED
  ) WITH (
     'connector' = 'filesystem', 
     'path' = 's3://some-bucket/output-table/',
     'format' = 'json',
     'sink.upsert-mode' = 'true'
   )
  

• 方案3：改用批处理专用TableEnvironment
  如果你的任务完全不需要流处理能力，可直接替换StreamTableEnvironment为批处理专用的TableEnvironment，默认运行在批模式下，删除输出表的主键配置即可正常运行。

额外优化提示：当前聚合逻辑中GROUP BY字段已经包含createdDate和id，max(createdDate)等价于分组内的createdDate，COUNT(DISTINCT(id))固定返回1，可根据实际业务需求调整聚合逻辑。

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