flink sql中如何设置 FlinkKafkaProducer.Semantic.UPSERT 参数

在 Flink SQL 中，可以通过使用 `CREATE TABLE` 语句创建 Kafka 表，并在 `WITH` 子句中指定 `value.format` 和 `sink.semantic` 参数来设置 `upsert` 语义。

例如，假设要将一个 Kafka 表写入到另外一个 Kafka 表中，并且只输出 `upsert` 数据，可以使用如下的 Flink SQL 语句：

CREATE TABLE output_kafka_table (
  id BIGINT,
  name STRING,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'output_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'value.format' = 'json',
  'sink.semantic' = 'upsert'
);

INSERT INTO output_kafka_table
SELECT id, name, age FROM input_kafka_table;

在上述语句中，`input_kafka_table` 是输入的 Kafka 表，`output_kafka_table` 是输出的 Kafka 表，`'value.format' = 'json'` 表示使用 JSON 格式序列化数据，`'sink.semantic' = 'upsert'` 表示只输出 `upsert` 数据。

需要注意的是，如果要使用 `upsert` 语义，输出的数据必须包含主键列，并且主键列的值不能为 `null`。在上述示例中，`output_kafka_table` 的主键是 `id` 列。如果输入的 Kafka 表中包含多个版本的数据，则只保留最新版本的数据，并且输出的数据也只包含最新版本的 `upsert` 数据。