flink1.1.36用SQL方式如何设置输出到hive表为upsert模式？

Tue, 05 Apr 2022 23:36:50 -0700 

问题：flink1.1.36用SQL方式如何设置输出到hive表为upsert模式？

flink：1.13.6
hive：1.1.1
hadoop：2.6.0-cdh5.16.2

纯SQL的方式，使用kafka作为source，中间的转换会有DISTINCT 或者 GROUP 操作，将计算结果sink到hive表，会报下面的错误

doesn't support consuming update changes which is produced by node 
GroupAggregate(groupBy=[id, user_id, status, EXPR$3]

在网上找了答案，说需要将sink表设置为upsert模式，尝试过按照下列方式创建sink表，创建表能成功，但提交INSERT INTO时还是报错

source表
 
CREATE TABLE data_2432_5074_model(
    id STRING,
    user_id STRING,status STRING
    ) WITH (
    'connector' = 'kafka',
    'topic' = 'person',
    'properties.bootstrap.servers' = '192.168.9.116:9092',
    'properties.group.id' = 'chinaoly-group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json',
    'json.fail-on-missing-field'='false',
    'json.ignore-parse-errors'='true'
)
sink表

CREATE TABLE output_2432_5076_model_1649226175146(
    id STRING,
    user_id STRING,
    status STRING,
    my_dt timestamp
) TBLPROPERTIES (
    'streaming-source.enable' = 'true',
    'streaming-source.partition.include' = 'all',
    'streaming-source.partition-order' = 'create-time',
    'sink.partition-commit.watermark-time-zone' = 'Asia/Shanghai',
    'sink.partition-commit.policy.kind' = 'metastore,success-file',
    'write.upsert.enable' = 'true',
    'streaming-source.monitor-interval' = '1 min'
)

计算逻辑
INSERT INTO output_2432_5076_model_1649226175146 SELECT DISTINCT id AS id, 
user_id AS user_id, status AS status ,proctime() FROM (SELECT * FROM 
data_2432_5074_model) WHERE status = '1'

万能的官方，能否给我答案，先谢谢了。



799590...@qq.com

回复