KSQL for aggregated historic data sorted by day with accepting late updates

Question

i've been trying to use ksql to aggregate data that comes in the following json pattern.

{"code": "1234", "type": "1234", "item": "1234", "items_sold": 3, "items_acquired": 0, "timestamp": "yyyyMMddhhmmssSSSSSSSSS"}

So far I've managed to aggregate the net change of items for any given day. I can also output the total current amount for any given item of type and code without problems.

The following is how i calculate net change for a given day

CREATE STREAM daily_change WITH (KAFKA_TOPIC='daily_change', PARTITIONS=1, REPLICAS=1) AS SELECT
  items_acquired - items_sold as net_change, 
  item, type, code,
  SUBSTRING(CAST(datets AS STRING), 0, 8) as curr_day
FROM change_event
EMIT CHANGES;

and from that stream i create a table

CREATE TABLE DAILY_INVENTORY WITH (KAFKA_TOPIC='daily_inventory', KEY_FORMAT='DELIMITED', PARTITIONS=1, REPLICAS=1, VALUE_FORMAT='JSON') AS SELECT
  SUM(net_change),
  curr_day, item, type, code
FROM daily_change
WINDOW TUMBLING ( SIZE 24 HOURS ) 
GROUP BY curr_day, item, type, code
EMIT CHANGES;

For the total inventory its the following

CREATE TABLE current_inventory WITH (KAFKA_TOPIC='current_inventory', KEY_FORMAT='DELIMITED', PARTITIONS=1, REPLICAS=1, VALUE_FORMAT='JSON') AS SELECT
  SUM(items_acquired - items_sold) current_stock,
  code, item, type
FROM change_event
GROUP BY code, item, type
EMIT CHANGES;

I'm stuck trying to get a table to show the total amount of inventory grouped by the day, code, item and type. On top of that problem, it also needs to accept changes of up to 14 days in the past.

The problem is that i have not managed to seperate the inventory by day. It either calculates only the net change of amount per day or the total current amount. Basically either i do not have the information of day or i do not have the information of total amount.

Some example data so i can illustrate the problem:

{"code": "1234", "type": "1234", "item": "1234", "items_sold": 0, "items_acquired": 2, "timestamp": "20220910113045000000000"} {"code": "1234", "type": "1234", "item": "1234", "items_sold": 1, "items_acquired": 0, "timestamp": "20220910143045000000000"}

At the end of the day this should result in a total of 1 for 20220910

{"code": "1234", "type": "1234", "item": "1234", "items_sold": 0, "items_acquired": 3, "timestamp": "20220912113045000000000"}

This should give me a total of 4 for 20220912

{"code": "1234", "type": "1234", "item": "1234", "items_sold": 0, "items_acquired": 1, "timestamp": "20220902113045000000000"}

This should adjust every total by 1, so 20220910 should now be 2, and 20220912 should now be 5.

I've tried using windowing with retention and grace and such, but haven't had success, as mentioned above i either end up producing the net_change or have to give up the information of the day timestamp

Does anybody have an idea what kind of approach i could try?

KSQL for aggregated historic data sorted by day with accepting late updates

0 Answers