物化视图与实时聚合

CREATE MATERIALIZED VIEW events_hourly_mv
TO events_hourly
AS SELECT
    toStartOfHour(event_time) AS hour,
    event_type,
    count() AS cnt,
    sum(value) AS total_value
FROM events
GROUP BY hour, event_type;

-- 目标表：按天分区的明细表
CREATE TABLE events_processed (
    event_time DateTime,
    user_id UInt64,
    event_type String,
    page_url String,
    processed_at DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_time, user_id);

-- 物化视图：清洗 + 加工
CREATE MATERIALIZED VIEW events_process_mv
TO events_processed
AS SELECT
    event_time,
    user_id,
    event_type,
    lower(page_url) AS page_url,      -- 清洗
    now() AS processed_at             -- 打上处理时间戳
FROM events;

-- 源表删除数据时，目标表不会自动同步
ALTER TABLE events DELETE WHERE event_type = 'spam';

-- 需要手动在目标表执行同样的清理
ALTER TABLE events_processed DELETE WHERE event_type = 'spam';

CREATE MATERIALIZED VIEW events_async_mv
TO events_hourly
AS SELECT
    toStartOfHour(event_time) AS hour,
    event_type,
    count() AS cnt
FROM events
GROUP BY hour, event_type
SETTINGS is_async = 1;   -- 24.10+ 实验特性

CREATE MATERIALIZED VIEW daily_summary_mv
REFRESH EVERY 1 HOUR
TO daily_summary
AS SELECT
    toDate(event_time) AS d,
    event_type,
    uniq(user_id) AS uu
FROM events
GROUP BY d, event_type;

SELECT name, status, last_refresh_start_time, last_refresh_success_time,
       last_refresh_duration_ms, error_count
FROM system.view_refreshes;

-- 目标表：保存聚合中间状态
CREATE TABLE events_agg (
    event_date Date,
    event_type String,
    cnt AggregateFunction(count, UInt64),
    uu AggregateFunction(uniq, UInt64),
    total AggregateFunction(sum, Float64)
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, event_type);

-- 物化视图：增量写入 -State 状态
CREATE MATERIALIZED VIEW events_agg_mv
TO events_agg
AS SELECT
    toDate(event_time) AS event_date,
    event_type,
    countState() AS cnt,
    uniqState(user_id) AS uu,
    sumState(value) AS total
FROM events
GROUP BY event_date, event_type;

-- 查询：用 -Merge 把状态还原成结果
SELECT
    event_date,
    event_type,
    countMerge(cnt) AS cnt,
    uniqMerge(uu) AS uu,
    sumMerge(total) AS total
FROM events_agg
GROUP BY event_date, event_type
ORDER BY event_date;

-- 第一层：原始明细 → 分钟聚合
CREATE TABLE agg_1min (... ) ENGINE = SummingMergeTree() ORDER BY (d, metric);
CREATE MATERIALIZED VIEW mv_1min TO agg_1min AS
SELECT toStartOfMinute(event_time) AS d, metric, countState() AS c
FROM raw_events GROUP BY d, metric;

-- 第二层：分钟聚合 → 小时聚合（读上一层的目标表）
CREATE TABLE agg_1hour (... ) ENGINE = SummingMergeTree() ORDER BY (d, metric);
CREATE MATERIALIZED VIEW mv_1hour TO agg_1hour AS
SELECT toStartOfHour(d) AS d, metric, sumMerge(c) AS c
FROM agg_1min GROUP BY d, metric;

SELECT
    table,
    sum(rows) AS total_rows,
    sum(bytes_on_disk) AS total_bytes,
    count() AS parts,
    max(modification_time) AS last_write
FROM system.parts
WHERE table IN ('events_hourly', 'events_agg') AND active = 1
GROUP BY table;

-- 对比源表与目标表的最大时间戳，判断视图是否滞后
SELECT
    'source' AS part, max(event_time) AS max_ts FROM events
UNION ALL
SELECT 'target', max(hour) FROM events_hourly;

-- system.tables 中能查到视图与目标表的关系（engine = MaterializedView 时）
SELECT database, name, engine, total_rows
FROM system.tables
WHERE engine = 'MaterializedView';

CREATE TABLE events_hourly (
    hour DateTime,
    event_type String,
    cnt UInt64
) ENGINE = SummingMergeTree()
ORDER BY (hour, event_type)
TTL hour + INTERVAL 30 DAY;   -- 30 天前的聚合结果自动清理

-- 错误示范：先插入目标表，又通过视图再插一次
INSERT INTO events_hourly SELECT toStartOfHour(now()), 'view', 1;
INSERT INTO events VALUES ('2024-06-01 10:00:00', 1, 'view', 1.5);
-- 同一个事件被 count 两次

-- POPULATE 会把源表已有历史数据一次性灌入视图
CREATE MATERIALIZED VIEW mv POPULATE TO target AS SELECT ... FROM source;

-- 安全的迁移顺序
DROP TABLE target;
DROP VIEW mv;
CREATE TABLE target (...);
CREATE MATERIALIZED VIEW mv TO target AS SELECT ... FROM source;
-- 最后手动回填历史数据
INSERT INTO target SELECT ... FROM source;