PyFlink使用UDTAGG写入Hive表报错不支持消费PythonGroupAggregate更新变更

• 你使用的普通Hive Parquet表Sink属于**仅追加写（Append Only）**类型的存储Sink，只能接收INSERT类型的变更数据。而你的计算逻辑用到了分组聚合（包括自定义UDTAGG表聚合函数），流模式下Flink的聚合算子默认会输出包含UPDATE/DELETE类型的变更日志（changelog流），普通Hive Sink不支持消费这类带更新的流，因此触发报错。
• Flink 1.13版本的原生Hive Sink默认未开启变更日志适配能力，且普通文件存储本身不支持随机修改已写入的数据，无法处理聚合产生的中间更新数据。

方案1：切换为批执行模式（离线批场景首选，改造成本最低） #

如果你的任务是离线批处理而非实时流处理，直接将运行模式改为批模式即可，批模式下聚合算子只会输出最终计算结果，不会产生中间更新的changelog：

# 配置批执行模式
table_env.get_config().set("execution.runtime-mode", "BATCH")

方案2：聚合结果转仅追加写入（流场景适用） #

如果是实时流任务，且你的业务逻辑是按分区全量覆盖写入，可以修改建表配置开启分区覆盖能力：

CREATE TABLE IF NOT EXISTS flink_tmp_model (
    run_id STRING,
    model_blob BINARY,
    roc_auc FLOAT
)  PARTITIONED BY (dt STRING) STORED AS parquet TBLPROPERTIES (
    'sink.partition-commit.delay'='1 s',
    'sink.partition-commit.policy.kind'='success-file',
    'sink.partition-commit.trigger'='partition-time',
    'auto-compaction'='true',
    'sink.rolling-policy.rollover-interval'='10 min',
    'execution.dynamic-table-options.enabled'='true'
)

写入时使用INSERT OVERWRITE语法指定覆盖对应分区即可：

INSERT OVERWRITE flink_tmp_model /*+ OPTIONS('sink.partition-commit.overwrite'='true') */
SELECT run_id, model_blob, roc_auc, dt FROM 你的聚合结果视图

方案3：更换支持更新的存储Sink（流场景需低延迟更新时适用） #

如果需要实时更新结果数据，可放弃普通Hive表作为结果表，改用支持行级更新的存储，比如HBase、JDBC数据库（MySQL/ClickHouse等）、Iceberg/Hudi这类支持更新的数据湖存储，这类Sink原生支持消费changelog的更新变更。

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