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;

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_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

--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 SOURCE kafka_twitch
FROM KAFKA BROKER 'kafka:9092' TOPIC 'twitch-streams'
  KEY FORMAT BYTES
  VALUE FORMAT BYTES
ENVELOPE UPSERT;

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

--What are the top10 games being played?
SELECT game_name, 
       cnt_streams, 
       agg_viewer_cnt 
FROM mv_agg_stream_game 
ORDER BY agg_viewer_cnt 
DESC LIMIT 10;

--Is anyone playing DOOM?
SELECT game_name, 

morsapaes

CREATE SOURCE rp_twitter_tweets
FROM KAFKA BROKER 'redpanda:9092' TOPIC 'dc_tweets'
  FORMAT BYTES;

morsapaes

CREATE MATERIALIZED VIEW twitter_tweets AS
SELECT (data->>'id')::bigint AS tweet_id,
       (data->'referenced_tweets'->0->>'type')::string AS tweet_type,
       (data->>'text')::string AS tweet_text,
       (data->'referenced_tweets'->0->>'id')::string AS tweet_id_rr,
       (data->>'author_id')::bigint AS user_id,
       (data->'geo'->>'place_id')::string AS place_id,
       (data->>'created_at')::timestamp AS created_at
FROM (SELECT CONVERT_FROM(data,'utf8')::jsonb AS data FROM rp_twitter_tweets);

morsapaes

CREATE VIEW twitter_tweets_enriched AS
SELECT tweet_text AS tweet,
	   username,
	   CASE WHEN tweet_type = 'quoted' THEN 'quoted retweet'
	        WHEN tweet_type = 'replied to' THEN 'tweet reply'
	   ELSE 'tweet'
	   END AS tweet_type,
       created_at
FROM twitter_tweets tt
--This is a streaming join!