Hi, Elakiya. Maybe this DDL could be executed. Please execute the select sql `select * from KafkaTable`. Then I think there will be some error or the `user_id` will not be read correctly.
Best, Hang elakiya udhayanan <laks....@gmail.com> 于2023年7月10日周一 16:25写道: > Hi Hang Ruan, > > Thanks for your response. But in the documentation, they have an example > of defining the Primary Key for the DDL statement (code below). In that > case we should be able to define the primary key for the DDL rite. We have > defined the primary key in our earlier use cases when it wasn't a nested > field. Please correct me , If I have misunderstood anything. > > CREATE TABLE KafkaTable ( `ts` TIMESTAMP(3) METADATA FROM 'timestamp', > `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING, PRIMARY KEY > (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'upsert-kafka', ... > 'key.format' = 'json', 'key.json.ignore-parse-errors' = 'true', > 'value.format' = 'json', 'value.json.fail-on-missing-field' = 'false', > 'value.fields-include' = 'EXCEPT_KEY') > > Thanks, > Elakiya > > On Mon, Jul 10, 2023 at 1:09 PM Hang Ruan <ruanhang1...@gmail.com> wrote: > >> 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 >>> >>