Streaming-ETL

Real-Time Inventory Management

You can use KSQL to update inventory in real time and keep track of the current available stock. This helps you plan accordingly and ensure efficient inventory management. Having just the right amount of inventory—not too much and not too little—based on the actual need is important for businesses that want to meet demand while minimizing costs.

The inventory data is stored in a topic (topic name : ‘inventory’). For this example, we are parsing JSON into an Apache Kafka® topic. This JSON has records that indicate when inventory is added (quantity (qty) is positive) and removed (qty is negative), simulating real-world inventory change events.

cat /Inventory.json | //bin/kafka-console-producer.sh --topic inventory --broker-list localhost:9092  --property "parse.key=true" --property "key.separator=:"

Sample inventory JSON:

1:{"cid":1,"item":"laptop","qty":10,"price":100}
2:{"cid":2,"item":"phone","qty":25,"price":98}
3:{"cid":3,"item":"watch","qty":100,"price":200}
4:{"cid":4,"item":"tablet","qty":20,"price":40}
1:{"cid":1,"item":"laptop","qty":10,"price":45}
2:{"cid":2,"item":"phone","qty":10,"price":210}
3:{"cid":3,"item":"watch","qty":-20,"price":195}
4:{"cid":4,"item":"tablet","qty":10,"price":38}

Now the inventory topic contains the inventory data.

Kafka Use Cases, KSQL
Environment4.1 or higher
Raj Sagiraju

Directions

1. Create a stream (inventory_stream) from the inventory topic to stream the inventory change events in real time.

CREATE STREAM inventory_stream (cid STRING, item STRING, qty INTEGER, price DOUBLE, balance INTEGER) with (VALUE_FORMAT='json',  KAFKA_TOPIC='inventory');

2. Now a table created from a 'Select' query on inventory_stream will have the most up-to-date inventory information by order of items. As new events arrive on the source inventory topic, the table will be updated.

SET 'auto.offset.reset' = 'earliest';
CREATE TABLE inventory_stream_table AS SELECT item, SUM(qty) AS item_qty FROM inventory_stream GROUP BY item;
Sample output :
ksql> SELECT item, item_qty from inventory_stream_table;
watch | 80
tablet | 30
laptop | 20
phone | 35
< Back to the Stream Processing Cookbook