Flink写clickhouse怎么实现精准一次性？

Clickhouse可以组合几种引擎达到幂等效果，ReplacingMergeTree + 物化视图 + AggregatingMergeTree(AggregateFunction:argMaxIf)

CREATETABLEIFNOTEXISTSdefault.local_dat_update ( pk1 UInt64 COMMENT '主键', pk2 String COMMENT '主键', ver DateTime64 COMMENT '版本,业务提供', col1 Int32 DEFAULT0 COMMENT '选取!=0的最大版本', col2 String DEFAULT'' COMMENT '选取!=空字符串的最大版本', col3 DateTimeDEFAULT0 COMMENT '选取!=0的最大版本' ) ENGINE = ReplacingMergeTree() PARTITIONBY (xxHash64(pk1, pk2) % 10) ORDER BY (pk1,pk2);

CREATETABLEIFNOTEXISTSdefault.local_agg_update ( pk1 UInt64, pk2 String, version AggregateFunction(max, DateTime64), col1 AggregateFunction(argMaxIf, Int32, DateTime64, UInt8), col2 AggregateFunction(argMaxIf, String, DateTime64, UInt8), col3 AggregateFunction(argMaxIf, DateTime, DateTime64, UInt8) ) ENGINE = AggregatingMergeTree() PARTITIONBY (xxHash64(pk1, pk2) % 10) ORDER BY (pk1, pk2);

CREATE MATERIALIZED VIEW IFNOTEXISTSdefault.mv_update TOdefault.local_agg_update AS SELECT pk1 AS pk1, pk2 AS pk2, maxState(ver) ASversion, -- 根据业务需求调整组合函数及判断逻辑 argMaxIfState(col1, ver, col1 !=0) AS col1, argMaxIfState(col2, ver, col2 !='') AS col2, argMaxIfState(col3, ver, col3 !=0) AS col3 FROMdefault.local_dat_update GROUP BY pk1, pk2;*来自志愿者整理的flink邮件归档