rodesai

DROP STREAM ALL_SALES;

rodesai

DROP STREAM ALL_SALES DELETE TOPIC;

rodesai

CREATE STREAM SALES_ONLINE (ITEMID BIGINT, CUSTOMERID BIGINT, PRICE INTEGER, PAYMENT_METHOD INTEGER, SHIPMENTID BIGINT) WITH (KAFKA_TOPIC='sales-online', VALUE_FORMAT='json');
CREATE STREAM SALES_INSTORE (ITEMID BIGINT, CUSTOMERID BIGINT, PRICE INTEGER, PAYMENT_METHOD INTEGER, STOREID BIGINT) WITH (KAFKA_TOPIC='sales-instore', VALUE_FORMAT='json');
CREATE STREAM ALL_SALES (ITEMID BIGINT, CUSTOMERID BIGINT, PRICE INTEGER, PAYMENT_METHOD INTEGER) WITH (KAFKA_TOPIC='sales', VALUE_FORMAT='json');
INSERT INTO ALL_SALES SELECT ITEMID, CUSTOMERID, PRICE, PAYMENT_METHOD FROM SALES_ONLINE;
INSERT INTO ALL_SALES SELECT ITEMID, CUSTOMERID, PRICE, PAYMENT_METHOD FROM SALES_INSTORE;
CREATE TABLE DAILY_SALES_PER_ITEM AS SELECT ITEMID, SUM(PRICE) FROM ALL_SALES WINDOW TUMBLING (SIZE 1 DAY) GROUP BY ITEMID;