iceberg flink upsert 流式数据入湖_iceberg upsert

-- flink v2

CREATE TABLE `sample_uniq_test01` (

  `id` BIGINT NOT NULL,

  `data` VARCHAR(2147483647),

  PRIMARY KEY (`id`) NOT ENFORCED

) PARTITIONED BY (`id`)

WITH (

  'format-version'='2',

  'catalog-database' = 'test001',

  'write.upsert.enabled' = 'true',

  'write.metadata.delete-after-commit.enabled' = 'true',

  'warehouse' = 'hdfs://nameservice2/user/hive/warehouse/',

  'uri' = 'thrift://10.8.49.114:9083,thrift://10.8.49.115:9083',

  'write.metadata.previous-versions-max' = '2',

  'catalog-table' = 'sample_uniq_test01',

  'catalog-type' = 'hive',

  'write.distribution-mode' = 'hash'

);

// 一定要有主键，且为主键在分区上，才可以的

insert into sample_uniq_test01 values (1,'1');

后面可以接入 flink kafka 表即可