Stream processing with KSQL

city_analytics.sql

CREATE STREAM city_analytics_realtime_full \
WITH (KAFKA_TOPIC='analytics.city_realtime_full',PARTITIONS=1,VALUE_FORMAT='JSON') AS \
select orderDate, \
     invoiceAddress->cityCode cityId, \
     ARRAY_LENGTH(lines) as cartCount , \
     orderStatus->name as status, \
     lines[1]->product->name productName, \
     lines[1]->product->productContentId contentId, \
     lines[1]->price productPrice, \
     lines[1]->imageUrls[1] productImage, \
     orderPrice->totalProduct totalProduct \
from analytics_created_order \
where orderStatus->name = 'Created' EMIT CHANGES;