Flink SQL中如何将嵌套字段设为Upsert-Kafka表主键？

问题根源 #

Flink SQL目前不支持将嵌套在ROW类型中的字段（比如employee.id）直接指定为表的主键，且Upsert-Kafka连接器要求主键必须对应Kafka消息的Key，无法直接从Value的嵌套结构中提取Key实现Upsert语义。

可行解决方案 #

方案1：将嵌套字段提取为顶层字段，直接定义主键

重新设计表结构，把employee.id和employee.name拆为表的顶层列，同时配置主键与Kafka Key的映射关系：

String statement = "CREATE TABLE Employee (\r\n" +
"  id STRING,\r\n" +
"  name STRING,\r\n" +
"  PRIMARY KEY (id) NOT ENFORCED\r\n" +
") WITH (\r\n" +
"  'connector' = 'upsert-kafka',\r\n" +
"  'topic' = 'employee',\r\n" +
"  'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',\r\n" +
"  'key.format' = 'raw',\r\n" +
"  'key.fields' = 'id',\r\n" +
"  'value.format' = 'avro-confluent',\r\n" +
"  'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n" +
"  'value.fields-include' = 'EXCEPT_KEY'\r\n" +
")";

• 注意：如果原Kafka Topic的消息Key并非id，需要先通过Flink作业将数据重新写入，把id作为Kafka Key；若原Topic的Key为空，该配置会自动将主键字段作为Kafka消息的Key写入。

方案2：先消费原始数据，再通过转换写入Upsert-Kafka表

如果无法直接修改解析结构，可分三步处理：

1. 创建普通Kafka表消费原始嵌套数据：

CREATE TABLE EmployeeRaw (
  employee ROW(id STRING, name STRING)
) WITH (
  'connector' = 'kafka',
  'topic' = 'employee',
  'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',
  'value.format' = 'avro-confluent',
  'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081'
);

2. 创建Upsert-Kafka目标表：

CREATE TABLE EmployeeUpsert (
  id STRING,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'employee-upsert',
  'properties.bootstrap.servers' = 'kafka-cp-kafka:9092',
  'key.format' = 'raw',
  'key.fields' = 'id',
  'value.format' = 'avro-confluent',
  'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081'
);

3. 插入数据并自动处理Upsert逻辑：

INSERT INTO EmployeeUpsert
SELECT employee.id, employee.name FROM EmployeeRaw;

关键说明 #

Upsert-Kafka的核心逻辑依赖Kafka消息的Key实现更新/插入语义，因此主键必须映射到Kafka的Key字段，而Flink SQL不支持嵌套字段作为主键，必须将其提升为顶层列。value.fields-include = 'EXCEPT_KEY'用于避免Value中重复存储Key字段，可根据业务需求调整为ALL。

内容的提问来源于stack exchange，提问作者JDP