Stream Aggregation

city_analytics_aggregation.sql

create table city_analytics_order_realtime_black_friday
with (KAFKA_TOPIC='city_analytics.order_realtime_black_friday',PARTITIONS=1,VALUE_FORMAT='JSON') as
select timestamptostring(orderDate, 'yyyy-MM') as orderDate,
cityId ,
SUM(cartCount) as totalCount,
SUM(productPrice) as totalPrice
from city_analytics_order_realtime_full where timestamptostring(orderDate, 'yyyy-MM-dd') >= '2020–11–27' and timestamptostring(orderDate, 'yyyy-MM-dd') <= '2020–11–30'
group by timestamptostring(orderDate, 'yyyy-MM'), cityId
EMIT CHANGES;