clickstream/ksql/ksql-clickstream-demo/demo/statements.sql

---------------------------------------------------------------------------------------------------
-- Create sources:
---------------------------------------------------------------------------------------------------
-- stream of user clicks:
CREATE STREAM clickstream (
        _time bigint,
        time varchar,
        ip varchar,
        request varchar,
        status int,
        userid int,
        bytes bigint,
        agent varchar
    ) with (
        kafka_topic = 'clickstream',
        value_format = 'json'
    );

-- error code lookup table:
CREATE TABLE clickstream_codes (
        code int primary key,
        definition varchar
    ) with (
        kafka_topic = 'clickstream_codes',
        value_format = 'json'
    );

-- users lookup table:
CREATE TABLE WEB_USERS (
        user_id int primary key,
        registered_At BIGINT,
        username varchar,
        first_name varchar,
        last_name varchar,
        city varchar,
        level varchar
    ) with (
        kafka_topic = 'clickstream_users',
        value_format = 'json'
    );

---------------------------------------------------------------------------------------------------
-- Build materialized stream views:
---------------------------------------------------------------------------------------------------

-- enrich click-stream with more user information:
CREATE STREAM USER_CLICKSTREAM AS
    SELECT
        userid,
        u.username,
        ip,
        u.city,
        request,
        status,
        bytes
    FROM clickstream c
        LEFT JOIN web_users u ON c.userid = u.user_id;

-- Enrich click-stream with more information about error codes:
CREATE STREAM ENRICHED_ERROR_CODES AS
    SELECT
        code,
        definition
    FROM clickstream
        LEFT JOIN clickstream_codes ON clickstream.status = clickstream_codes.code;

---------------------------------------------------------------------------------------------------
-- Build materialized table views:
---------------------------------------------------------------------------------------------------

-- Per-userId tables -----------------------------------------------------------------------------

-- Table of events per minute for each user:
CREATE table events_per_min AS
    SELECT
        userid as k1,
        AS_VALUE(userid) as userid,
        WINDOWSTART as EVENT_TS,
        count(*) AS events
    FROM clickstream window TUMBLING (size 60 second)
    GROUP BY userid;

-- Table of html pages per minute for each user:
CREATE TABLE pages_per_min AS
    SELECT
        userid as k1,
        AS_VALUE(userid) as userid,
        WINDOWSTART as EVENT_TS,
        count(*) AS pages
    FROM clickstream WINDOW HOPPING (size 60 second, advance by 5 second)
    WHERE request like '%html%'
    GROUP BY userid;

-- Per-username tables ----------------------------------------------------------------------------

-- User sessions table - 30 seconds of inactivity expires the session
-- Table counts number of events within the session
CREATE TABLE CLICK_USER_SESSIONS AS
    SELECT
        username as K,
        AS_VALUE(username) as username,
        WINDOWEND as EVENT_TS,
        count(*) AS events
    FROM USER_CLICKSTREAM window SESSION (30 second)
    GROUP BY username;

-- Per-status tables ------------------------------------------------------------------------------

-- number of errors per min, using 'HAVING' Filter to show ERROR codes > 400 where count > 5
CREATE TABLE ERRORS_PER_MIN_ALERT AS
    SELECT
        status as k1,
        AS_VALUE(status) as status,
        WINDOWSTART as EVENT_TS,
        count(*) AS errors
    FROM clickstream window HOPPING (size 60 second, advance by 20 second)
    WHERE status > 400
    GROUP BY status
    HAVING count(*) > 5 AND count(*) is not NULL;

-- number of errors per min
CREATE TABLE ERRORS_PER_MIN AS
    SELECT
        status as k1,
        AS_VALUE(status) as status,
        WINDOWSTART as EVENT_TS,
        count(*) AS errors
    FROM clickstream window HOPPING (size 60 second, advance by 5  second)
    WHERE status > 400
    GROUP BY status;

-- Per-error code tables --------------------------------------------------------------------------

-- Enriched error codes table:
-- Aggregate (count&groupBy) using a TABLE-Window
CREATE TABLE ENRICHED_ERROR_CODES_COUNT WITH (key_format='json') AS
    SELECT
        code as k1,
        definition as K2,
        AS_VALUE(code) as code,
        WINDOWSTART as EVENT_TS,
        AS_VALUE(definition) as definition,
        COUNT(*) AS count
    FROM ENRICHED_ERROR_CODES WINDOW TUMBLING (size 30 second)
    GROUP BY code, definition
    HAVING COUNT(*) > 1;

-- Enriched user details table:
-- Aggregate (count&groupBy) using a TABLE-Window
CREATE TABLE USER_IP_ACTIVITY WITH (key_format='json') AS
    SELECT
        username as k1,
        ip as k2,
        city as k3,
        AS_VALUE(username) as username,
        WINDOWSTART as EVENT_TS,
        AS_VALUE(ip) as ip,
        AS_VALUE(city) as city,
        COUNT(*) AS count
    FROM USER_CLICKSTREAM WINDOW TUMBLING (size 60 second)
    GROUP BY username, ip, city
    HAVING COUNT(*) > 1;