Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic，支持做数据源表和结果表。

• 作为源表时，Upsert Kafka 连接器可以将 Kafka 中存储的数据转换为 changelog 流，其中每条数据记录代表一个更新或删除事件。数据记录中有 key，表示 UPDATE；数据记录中没有 key，表示 INSERT；数据记录中 key 的 value 为空，表示 DELETE。
• 作为结果表时，Upsert Kafka 连接器可以消费上游计算逻辑产生的 changelog 流。它会将 INSERT 或 UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入，表示对应 key 的消息被删除。
  Flink将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新或删除消息将落在同一分区中。

使用限制

Upsert-kafka 连接器暂时仅支持在 Flink 1.16-volcano 引擎版本中使用。

DDL 定义

CREATE TABLE upsert_kafka_sink (
  user_region STRING,
  pv BIGINT,
  uv BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = '<yourTopicName>',
  'properties.bootstrap.servers' = '...',
  'key.format' = 'avro',
  'value.format' = 'avro'
);

WITH 参数

示例代码

• 源表

  CREATE TABLE upsert_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp,
       PRIMARY KEY (order_id) NOT ENFORCED
       )
       WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'source',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'group_01', 
    'key.format' = 'json', 
    'value.format' = 'json'  
  );
  
   CREATE TABLE print_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp
      )
  WITH (
       'connector' = 'print'           
  );
  
  insert into print_sink
  select * from upsert_source; 
  

• 结果表

  CREATE TABLE datagen_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time as localtimestamp
      )
  WITH (
       'connector' = 'datagen',
       'rows-per-second' = '5'          
  );
  
  CREATE TABLE upsert_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp,
       PRIMARY KEY (order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'sink',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'group_01', 
    'key.format' = 'json', 
    'value.format' = 'json'  
  );
  
  
  INSERT INTO upsert_sink
  SELECT * FROM datagen_source;