morsapaes

CREATE MATERIALIZED SOURCE mz_source 
FROM POSTGRES
  CONNECTION 'host=postgres port=5432 user=postgres dbname=postgres password=postgres'
  PUBLICATION 'mz_source';

CREATE VIEWS FROM SOURCE mz_source (stream_tag_ids);

morsapaes

CREATE SOURCE kafka_twitch
FROM KAFKA BROKER 'kafka:9092' TOPIC 'twitch-streams'
  KEY FORMAT BYTES
  VALUE FORMAT BYTES
ENVELOPE UPSERT;

morsapaes

--Use a view to create a "shortcut" 
--to the top10 games query
CREATE VIEW v_stream_game_top10 AS 
SELECT game_id, 
       game_name, 
       agg_viewer_cnt 
FROM mv_agg_stream_game 
ORDER BY agg_viewer_cnt DESC 
LIMIT 10;

morsapaes

CREATE MATERIALIZED VIEW mv_agg_stream_tag AS
SELECT st.localization_name AS tag,
       cnt_tag
FROM (
       --Expand the tags array for each event
       --and aggregate before joining
       SELECT tg, COUNT(*) AS cnt_tag
       FROM v_twitch_stream ts,
            unnest(tag_ids) tg 
       WHERE game_id IS NOT NULL

morsapaes

CREATE MATERIALIZED VIEW mv_stream_15min AS
SELECT title,
       user_name,
       game_name,
       started_at
FROM v_twitch_stream
WHERE game_id IS NOT NULL
  --Use temporal filters to filter events with 
  --a sliding window of 15 minutes (900000 miliseconds)
  AND (mz_logical_timestamp() >= (extract('epoch' from started_at)*1000)::bigint 

morsapaes

CREATE MATERIALIZED VIEW mv_agg_stream_game AS
SELECT game_id,
       game_name,
       COUNT(id) AS cnt_streams,
       SUM(viewer_count) AS agg_viewer_cnt
FROM v_twitch_stream
WHERE game_id IS NOT NULL
GROUP BY game_id, game_name;