Hi, elakiya.
The upsert-kafka connector will read the primary keys from the Kafka
message keys. We cannot define the fields in the Kafka message values as
the primary key.
Best,
Hang
elakiya udhayanan <laks....@gmail.com> 于2023年7月10日周一 13:49写道：
> Hi team,
>
> I have a Kafka topic named employee which uses confluent avro schema and
> will emit the payload as below:
>
> {
> "employee": {
> "id": "123456",
> "name": "sampleName"
> }
> }
> I am using the upsert-kafka connector to consume the events from the above
> Kafka topic as below using the Flink SQL DDL statement, also here I want to
> use the id field as the Primary key. But I am unable to use the id field
> since it is inside the object.
>
> DDL Statement:
> String statement = "CREATE TABLE Employee (\r\n" +
> " employee ROW(id STRING, name STRING\r\n" +
> " ),\r\n" +
> " PRIMARY KEY (employee.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" +
> " 'value.format' = 'avro-confluent',\r\n" +
> " 'value.avro-confluent.url' = 'http://kafka-cp-schema-registry:8081',\r\n"
> +
> ")";
> Any help is appreciated TIA
>
> Thanks,
> Elakiya
>
