KSQL STRUCT Fields For TIMESTAMP

Question

is there are a way to make this work

JSON DATA

"Header": {
    "StoreID": 10225,
    "BusinessDate": "2019-05-03",
    "PeriodBusinessDate": "2019-05-03",
    "ProcessMode": "Partial"
  }

I try this but is give me : No column with the provided timestamp column name in the WITH clause, HEADER->BUSINESSDATE, exists in the defined schema.

CREATE STREAM test2 (HEADER STRUCT<StoreID int,BusinessDate VARCHAR>) WITH (KAFKA_TOPIC='hermes__output__tfrema__v1',VALUE_FORMAT='JSON',
timestamp='HEADER->BusinessDate',timestamp_format='yyyy-MMM-dd');

Answer 1

You can't use nested fields in the TIMESTAMP parameter. You'd need to extract it first and then use it. For example:

CREATE STREAM X (COL1 INT, COL2 VARCHAR, HEADER STRUCT<StoreID int,BusinessDate VARCHAR>) 
  WITH (KAFKA_TOPIC='hermes__output__tfrema__v1',VALUE_FORMAT='JSON')

CREATE STREAM Y AS 
  SELECT COL1, COL2, HEADER->BusinessDate AS BusinessDate, HEADER 
  FROM X;

CREATE STREAM Z COL1 INT, COL2 VARCHAR, BusinessDate VARCHAR, HEADER STRUCT<StoreID int,BusinessDate VARCHAR>) 
  WITH (KAFKA_TOPIC='Y',VALUE_FORMAT='JSON',timestamp='BusinessDate',timestamp_format='yyyy-MMM-dd');)

If you're using Avro you can simplify things because the schema wouldn't need restating:

CREATE STREAM X (COL1 INT, COL2 VARCHAR, HEADER STRUCT<StoreID int,BusinessDate VARCHAR>) 
  WITH (KAFKA_TOPIC='hermes__output__tfrema__v1',VALUE_FORMAT='JSON')

CREATE STREAM Y WITH (VALUE_FORMAT='AVRO') 
  AS SELECT COL1, COL2, HEADER->BusinessDate, HEADER FROM X;

CREATE STREAM Z 
  WITH (KAFKA_TOPIC='Y',VALUE_FORMAT='JSON',timestamp='BusinessDate',timestamp_format='yyyy-MMM-dd');)