ksql create stream from json key/value

Viewed 389

I have a problem with apache kafka and the output of connector. when i try to create a stream from the topic i get some errors.

the data of the topic is this one ( without a schema, in json format ):

key:

{
  "payload": {
    "sourceName": "HotPump",
    "jobName": "pollingHotPump"
  }
}

value:

{
  "payload": {
    "fields": {
      "cw": [
        4657,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        13108,
        16637,
        0,
        0,
        0
      ]
    },
    "timestamp": 1638540457655,
    "expires": null,
    "connection-name": "Condensator"
  }
}

the ksql query for create a stream is this one:


 CREATE STREAM s_devices
  ( 
    key struct<payload struct<sourceName string>> ,
    value struct<payload struct<fields struct<cw array>>>,
    ts struct<payload struct<timestamp bigint>>

   )
  WITH (KAFKA_TOPIC='devices',
        VALUE_FORMAT='JSON', KEY_FORMAT='JSON');

The result of ksql client is: "Failed to prepare statement: Cannot resolve unknown type: ARRAY"

When i try to create the stream only with key struct<payload struct<sourceName string>> the select select key->payload->sourceName, value->payload->timestamp from s_devices; working correctly and the value is showed;

when i try only with ts struct<payload struct<timestamp bigint>> the table is created but when i try to select the value is null select value->payload->timestamp from s_devices;

where is the error?

thanks

0 Answers
Related