How to work with datetime+offset in streams

ksqlDB
1

Hi all.

Have a column/value pair coming in as follows (Json message so it’s first seen/interprided as a varchar, the schema registry i defined it as type string, format date)…

"PayDateTime": "2023-12-12-T13:30:37+02:00",

want to cast it into a date/time/timestamp column in streams.

please help.

G

2

… could not figure out how to work with a string time that includes GMT offset so left that as is… added a string field that contained the unix epoc value and then defined that via the ksql as a timestamp with desired results

Curious to know, the schema i define the field as a number, but it’s getting a string posted to it, topic is happily accepting the payloads?

Schema

{
  "$schema": "http://json-schema.org/draft-04/schema#",
  "properties": {
    "BasketItems": {
      "items": [
        {
          "properties": {
            "Brand": {
              "type": "string"
            },
            "Catergory": {
              "type": "string"
            },
            "Id": {
              "type": "string"
            },
            "Name": {
              "type": "string"
            },
            "Price": {
              "type": "number"
            },
            "Quantity": {
              "type": "integer"
            }
          },
          "required": [
            "Id",
            "Name",
            "Brand",
            "Catergory",
            "Price",
            "Quantity"
          ],
          "type": "object"
        }
      ],
      "type": "array"
    },
    "Clerk": {
      "properties": {
        "Id": {
          "type": "string"
        },
        "Name": {
          "type": "string"
        }
      },
      "required": [
        "Id",
        "Name"
      ],
      "type": "object"
    },
    "InvoiceNumber": {
      "type": "string"
    },
    "Nett": {
      "type": "number"
    },
    "SaleDateTime": {
      "format": "date",
      "type": "string"
    },
    "SaleTimestamp": {
      "type": "number"
    },
    "Store": {
      "properties": {
        "Id": {
          "type": "string"
        },
        "Name": {
          "type": "string"
        }
      },
      "required": [
        "Id",
        "Name"
      ],
      "type": "object"
    },
    "TerminalPoint": {
      "type": "string"
    },
    "Total": {
      "type": "number"
    },
    "VAT": {
      "type": "number"
    }
  },
  "required": [
    "InvoiceNumber",
    "SaleDateTime",
    "SaleTimestamp",
    "Store",
    "Clerk",
    "TerminalPoint",
    "BasketItems",
    "Nett",
    "VAT",
    "Total"
  ],
  "type": "object"
}

Stream Create

CREATE STREAM salesbasket (
	   	InvoiceNumber VARCHAR ,
	 	SaleDateTime VARCHAR,
	 	SaleTimestamp TIMESTAMP,
	  	TerminalPoint VARCHAR,
	   	Nett DOUBLE,
	  	Vat DOUBLE,
	 	Total DOUBLE,
       	Store STRUCT<
       		Id VARCHAR,
     		Name VARCHAR>,
     	Clerk STRUCT<
     		Id VARCHAR,
          	Name VARCHAR>,
    	BasketItems ARRAY< STRUCT<id VARCHAR,
        	Name VARCHAR,
          	Brand VARCHAR,
          	Catergory VARCHAR,
         	Price DOUBLE,
        	Quantity integer >>) 
WITH (KAFKA_TOPIC='salesbasket',
		VALUE_FORMAT='JSON',
       	PARTITIONS=1);
3

My solution in the end.

I added a unix/epoc based timestamp (SaleTimestamp) as a string of numbers to the payload, (my original SaleDateTime is a string, that includes GMT Offset),

I’ve updated the schema registry to also see this new field… although in schema registry it’s defined as a number, but it’s passed in from source app as a string. Strangely the payload is accepted.

G

4

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.