ClickHouse 实时分析场景

CREATE TABLE user_events (
    event_time DateTime,
    user_id UInt64,
    session_id UInt64,
    event_type LowCardinality(String),
    -- 事件类型：page_view, product_view, add_to_cart, checkout, payment
    page_url String,
    product_id UInt64,
    platform LowCardinality(String),  -- ios, android, web
    version String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, event_time, user_id);

-- 分析从浏览到支付的转化率
WITH steps AS (
    SELECT
        user_id,
        session_id,
        groupArray(event_type) AS event_sequence,
        groupArray(event_time) AS time_sequence
    FROM user_events
    WHERE event_time > now() - INTERVAL 7 DAY
    GROUP BY user_id, session_id
    HAVING has(event_sequence, 'page_view')
)
SELECT
    countIf(has(event_sequence, 'page_view')) AS step1_views,
    countIf(has(event_sequence, 'product_view')) AS step2_product,
    countIf(has(event_sequence, 'add_to_cart')) AS step3_cart,
    countIf(has(event_sequence, 'checkout')) AS step4_checkout,
    countIf(has(event_sequence, 'payment')) AS step5_payment,
    round(step2_product / step1_views * 100, 2) AS step1_to_2,
    round(step5_payment / step1_views * 100, 2) AS overall_conversion
FROM steps;

-- 检查用户在 30 分钟内是否按顺序完成各步骤
WITH 
-- 找到各用户在每个步骤的首次发生时间
first_events AS (
    SELECT
        user_id,
        event_type,
        min(event_time) AS first_time
    FROM user_events
    WHERE event_time > now() - INTERVAL 7 DAY
      AND event_type IN ('page_view', 'add_to_cart', 'payment')
    GROUP BY user_id, event_type
),
-- 将各步骤时间 pivot 出来
funnel AS (
    SELECT
        user_id,
        maxIf(first_time, event_type = 'page_view') AS t1,
        maxIf(first_time, event_type = 'add_to_cart') AS t2,
        maxIf(first_time, event_type = 'payment') AS t3
    FROM first_events
    GROUP BY user_id
)
SELECT
    count() AS total_users,
    countIf(t1 IS NOT NULL) AS step1,
    countIf(t2 > t1 AND t2 <= t1 + INTERVAL 30 MINUTE) AS step2_in_window,
    countIf(t3 > t2 AND t3 <= t2 + INTERVAL 30 MINUTE) AS step3_in_window,
    round(step2_in_window / step1 * 100, 2) AS step1_2_rate,
    round(step3_in_window / step1 * 100, 2) AS step1_3_rate
FROM funnel;

CREATE TABLE user_activity (
    activity_date Date,
    user_id UInt64,
    -- 使用 bitmap 存储当天的行为编码
    -- 可以进一步扩展存储详细行为
    sign UInt8  -- 1 表示当天活跃
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(activity_date)
ORDER BY (activity_date, user_id);

-- 计算 N 日留存率
WITH 
-- 各用户首次活跃日期
cohorts AS (
    SELECT
        user_id,
        min(activity_date) AS first_active
    FROM user_activity
    GROUP BY user_id
),
-- 各用户在首次活跃后的每日活跃情况
retention AS (
    SELECT
        c.user_id,
        c.first_active,
        a.activity_date,
        dateDiff('day', c.first_active, a.activity_date) AS day_offset
    FROM cohorts c
    JOIN user_activity a ON c.user_id = a.user_id
)
SELECT
    first_active AS cohort_date,
    count(DISTINCT user_id) AS cohort_size,
    countDistinctIf(user_id, day_offset = 0) AS d0,
    countDistinctIf(user_id, day_offset = 1) AS d1,
    countDistinctIf(user_id, day_offset = 7) AS d7,
    countDistinctIf(user_id, day_offset = 30) AS d30,
    round(d1 / cohort_size * 100, 2) AS day1_retention,
    round(d7 / cohort_size * 100, 2) AS day7_retention,
    round(d30 / cohort_size * 100, 2) AS day30_retention
FROM retention
GROUP BY first_active
ORDER BY first_active DESC
LIMIT 30;

-- 分析用户连续活跃天数
WITH user_streaks AS (
    SELECT
        user_id,
        activity_date,
        activity_date - INTERVAL row_number() OVER (PARTITION BY user_id ORDER BY activity_date) DAY AS streak_group
    FROM user_activity
    WHERE sign = 1
),
streak_lengths AS (
    SELECT
        user_id,
        streak_group,
        count() AS streak_length
    FROM user_streaks
    GROUP BY user_id, streak_group
)
SELECT
    streak_length,
    count(DISTINCT user_id) AS users
FROM streak_lengths
GROUP BY streak_length
ORDER BY streak_length;

CREATE TABLE ab_test_events (
    event_time DateTime,
    user_id UInt64,
    experiment_id String,
    variant LowCardinality(String),  -- 'control' 或 'treatment'
    event_type LowCardinality(String),  -- 'page_view', 'click', 'conversion'
    value Float64 DEFAULT 0
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (experiment_id, variant, event_time);

WITH metrics AS (
    SELECT
        experiment_id,
        variant,
        count(DISTINCT user_id) AS users,
        countIf(event_type = 'page_view') AS page_views,
        countIf(event_type = 'conversion') AS conversions,
        sumIf(value, event_type = 'purchase') AS revenue
    FROM ab_test_events
    WHERE event_time > now() - INTERVAL 7 DAY
    GROUP BY experiment_id, variant
)
SELECT
    experiment_id,
    maxIf(users, variant = 'control') AS control_users,
    maxIf(users, variant = 'treatment') AS treatment_users,
    round(maxIf(conversions, variant = 'treatment') / maxIf(users, variant = 'treatment') * 100, 3) AS treatment_conversion,
    round(maxIf(conversions, variant = 'control') / maxIf(users, variant = 'control') * 100, 3) AS control_conversion,
    round(
        (treatment_conversion - control_conversion) / control_conversion * 100,
        2
    ) AS lift_pct
FROM metrics
GROUP BY experiment_id;

CREATE TABLE user_profiles (
    user_id UInt64,
    -- 人口属性
    gender LowCardinality(String),
    age_group LowCardinality(String),
    city LowCardinality(String),
    
    -- 行为标签（用位运算编码或使用数组）
    interests Array(LowCardinality(String)),
    
    -- 计算指标
    first_seen DateTime,
    last_seen DateTime,
    total_orders UInt32,
    lifetime_value Float64,
    
    -- 更新版本号（用于 ReplacingMergeTree）
    version UInt64
) ENGINE = ReplacingMergeTree(version)
ORDER BY user_id;

-- 查询特定用户画像
SELECT
    age_group,
    gender,
    city,
    interests,
    lifetime_value
FROM user_profiles FINAL
WHERE user_id = 12345;

-- RFM 分析（Recency, Frequency, Monetary）
WITH user_rfm AS (
    SELECT
        user_id,
        dateDiff('day', max(order_time), today()) AS recency,
        count() AS frequency,
        sum(amount) AS monetary
    FROM orders
    WHERE order_time > now() - INTERVAL 1 YEAR
    GROUP BY user_id
),
rfm_scored AS (
    SELECT
        user_id,
        recency,
        frequency,
        monetary,
        -- RFM 评分（1-5分）
        ntile(5) OVER (ORDER BY recency DESC) AS r_score,
        ntile(5) OVER (ORDER BY frequency) AS f_score,
        ntile(5) OVER (ORDER BY monetary) AS m_score
    FROM user_rfm
)
SELECT
    r_score,
    f_score,
    m_score,
    count() AS users,
    CASE
        WHEN r_score >= 4 AND f_score >= 4 THEN 'Champions'
        WHEN r_score >= 3 AND f_score >= 3 THEN 'Loyal Customers'
        WHEN r_score >= 4 AND f_score <= 2 THEN 'New Customers'
        WHEN r_score <= 2 AND f_score >= 3 THEN 'At Risk'
        ELSE 'Others'
    END AS segment
FROM rfm_scored
GROUP BY r_score, f_score, m_score
ORDER BY users DESC;

CREATE TABLE user_profile_wide (
    user_id UInt64,
    -- 高频标签直接存储，便于查询和索引
    age_group LowCardinality(String),
    gender LowCardinality(String),
    city_level UInt8,  -- 1-5线城市
    device_type LowCardinality(String),
    
    -- 低频标签聚合为 Map，减少列数
    tags Map(LowCardinality(String), String),
    
    -- 行为统计指标
    dau_count_7d UInt16,      -- 近7天活跃天数
    dau_count_30d UInt16,     -- 近30天活跃天数
    total_orders UInt32,
    last_order_time DateTime,
    
    updated_at DateTime
) ENGINE = ReplacingMergeTree(updated_at)
ORDER BY user_id;

-- 通过物化视图自动更新画像
CREATE MATERIALIZED VIEW user_profile_mv TO user_profile_wide AS
SELECT
    user_id,
    any(age_group) AS age_group,
    any(gender) AS gender,
    city_level,
    device_type,
    mapFromArrays(
        ['interest_sports', 'interest_tech', 'interest_finance'],
        [toString(has_interest_sports), toString(has_interest_tech), toString(has_interest_finance)]
    ) AS tags,
    countIf(event_date > today() - 7) AS dau_count_7d,
    countIf(event_date > today() - 30) AS dau_count_30d,
    total_orders,
    last_order_time,
    now() AS updated_at
FROM user_events_raw
GROUP BY user_id, city_level, device_type;

-- 将秒级数据降采样为分钟级
SELECT
    toStartOfMinute(event_time) AS minute,
    avg(value) AS avg_value,
    max(value) AS max_value,
    min(value) AS min_value,
    count() AS sample_count
FROM sensor_readings
WHERE event_time > now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute;

-- 使用物化视图自动降采样
CREATE TABLE sensor_minute (
    minute DateTime,
    sensor_id UInt64,
    avg_value Float64,
    max_value Float64,
    min_value Float64,
    sample_count UInt64
) ENGINE = MergeTree()
ORDER BY (sensor_id, minute);

CREATE MATERIALIZED VIEW sensor_minute_mv TO sensor_minute AS
SELECT
    toStartOfMinute(event_time) AS minute,
    sensor_id,
    avg(value) AS avg_value,
    max(value) AS max_value,
    min(value) AS min_value,
    count() AS sample_count
FROM sensor_readings
GROUP BY sensor_id, minute;

-- 基于移动平均的异常检测
WITH stats AS (
    SELECT
        sensor_id,
        value,
        event_time,
        avg(value) OVER (
            PARTITION BY sensor_id
            ORDER BY event_time
            RANGE BETWEEN INTERVAL 5 MINUTE PRECEDING AND CURRENT ROW
        ) AS moving_avg,
        stddevSamp(value) OVER (
            PARTITION BY sensor_id
            ORDER BY event_time
            RANGE BETWEEN INTERVAL 5 MINUTE PRECEDING AND CURRENT ROW
        ) AS moving_std
    FROM sensor_readings
    WHERE event_time > now() - INTERVAL 1 HOUR
)
SELECT
    sensor_id,
    event_time,
    value,
    moving_avg,
    abs(value - moving_avg) / moving_std AS z_score
FROM stats
WHERE abs(value - moving_avg) / moving_std > 3  -- Z-score > 3 为异常
ORDER BY event_time DESC;

-- 分析用户在同一会话内的页面跳转路径
WITH session_pages AS (
    SELECT
        user_id,
        session_id,
        groupArray(page_url) AS page_sequence,
        groupArray(event_time) AS time_sequence
    FROM user_events
    WHERE event_type = 'page_view'
      AND event_time > now() - INTERVAL 1 DAY
    GROUP BY user_id, session_id
    HAVING length(page_sequence) > 1
)
SELECT
    page_sequence[1] AS landing_page,
    page_sequence[2] AS second_page,
    count() AS session_count,
    avg(length(page_sequence)) AS avg_pages_per_session
FROM session_pages
GROUP BY landing_page, second_page
ORDER BY session_count DESC
LIMIT 20;

-- 扩展同期群分析：计算各群的月留存矩阵
WITH cohorts AS (
    SELECT
        user_id,
        min(toStartOfMonth(activity_date)) AS cohort_month
    FROM user_activity
    GROUP BY user_id
),
activity AS (
    SELECT DISTINCT
        c.user_id,
        c.cohort_month,
        a.activity_month,
        dateDiff('month', c.cohort_month, a.activity_month) AS month_offset
    FROM cohorts c
    JOIN (
        SELECT user_id, toStartOfMonth(activity_date) AS activity_month
        FROM user_activity
        GROUP BY user_id, activity_month
    ) a ON c.user_id = a.user_id
),
cohort_sizes AS (
    SELECT cohort_month, count() AS cohort_size FROM cohorts GROUP BY cohort_month
)
SELECT
    c.cohort_month,
    s.cohort_size,
    sumIf(cohort_size, month_offset = 0) AS m0,
    sumIf(cohort_size, month_offset = 1) AS m1,
    sumIf(cohort_size, month_offset = 2) AS m2,
    sumIf(cohort_size, month_offset = 3) AS m3,
    round(sumIf(cohort_size, month_offset = 1) / s.cohort_size * 100, 1) AS m1_retention,
    round(sumIf(cohort_size, month_offset = 3) / s.cohort_size * 100, 1) AS m3_retention
FROM activity c
JOIN cohort_sizes s ON c.cohort_month = s.cohort_month
GROUP BY c.cohort_month, s.cohort_size
ORDER BY c.cohort_month DESC
LIMIT 12;