时序分析最佳实践：时间序列建模、降采样与异常检测 SQL

CREATE TABLE metrics (
    ts DateTime64(3),
    host String,
    metric String,
    value Float64
) ENGINE = MergeTree()
PARTITION BY toDate(ts)               -- 按天分区，利于 TTL/裁剪
ORDER BY (metric, host, ts)           -- 排序键：先维度后时间
TTL toDate(ts) + INTERVAL 30 DAY;     -- 自动清理 30 天前数据

-- toStartOfInterval（推荐，支持任意间隔）
SELECT
    toStartOfInterval(ts, INTERVAL 5 MINUTE) AS bucket,
    count() AS cnt,
    avg(value) AS avg_val
FROM metrics
WHERE metric = 'cpu'
GROUP BY bucket
ORDER BY bucket;

-- dateBin（对 Date 类型）
SELECT dateBin(INTERVAL 1 DAY, ts, toDateTime('2020-01-01')) AS day, ...

-- 环比：今天的值 vs 昨天的值
WITH today AS (
    SELECT toStartOfInterval(ts, INTERVAL 1 HOUR) AS h, avg(value) AS v
    FROM metrics WHERE metric = 'cpu' GROUP BY h
)
SELECT
    t.h,
    t.v AS today,
    y.v AS yesterday,
    (t.v - y.v) / y.v * 100 AS pct_change
FROM today t
LEFT JOIN today y ON y.h = t.h - INTERVAL 1 HOUR;

-- 每行取"过去 1 小时"的累计值（窗口函数）
SELECT
    ts,
    sum(value) OVER (
        PARTITION BY host
        ORDER BY ts
        ROWS BETWEEN 60 PRECEDING AND CURRENT ROW
    ) AS rolling_sum
FROM metrics
WHERE metric = 'requests';

-- 原始数据 → 5 分钟聚合表（长期保留）
CREATE MATERIALIZED VIEW metrics_5m
ENGINE = AggregatingMergeTree()
ORDER BY (metric, host, bucket)
AS SELECT
    toStartOfInterval(ts, INTERVAL 5 MINUTE) AS bucket,
    metric, host,
    avgState(value) AS avg_value,
    maxState(value) AS max_value,
    minState(value) AS min_value
FROM metrics
GROUP BY bucket, metric, host;

SELECT bucket, avgMerge(avg_value) AS avg_val
FROM metrics_5m
WHERE metric = 'cpu'
ORDER BY bucket;

-- 原始表 TTL：7 天过期（明细）
TTL toDate(ts) + INTERVAL 7 DAY;
-- 降采样表单独保留 30 天

-- 值的变化量（相邻采样差）
SELECT
    ts,
    value,
    value - lagInFrame(value) OVER (ORDER BY ts) AS delta
FROM metrics
WHERE metric = 'cpu'
ORDER BY ts;

-- 变化率
SELECT
    ts,
    value,
    (value - lagInFrame(value) OVER (ORDER BY ts))
      / lagInFrame(value) OVER (ORDER BY ts) * 100 AS pct_delta
FROM metrics
ORDER BY ts;

-- 用过去 1 小时均值作为基线，当前值偏离多少倍标准差
WITH stats AS (
    SELECT
        avg(value) AS mu,
        stddevSamp(value) AS sigma,
        now() - INTERVAL 1 HOUR AS win_start
    FROM metrics
    WHERE metric = 'cpu' AND ts >= now() - INTERVAL 1 HOUR
)
SELECT
    m.ts, m.value,
    (m.value - s.mu) / s.sigma AS zscore
FROM metrics m, stats s
WHERE m.metric = 'cpu'
  AND m.ts >= now()
ORDER BY m.ts;

-- 找出超过 3σ 的异常点
WITH base AS (
    SELECT ts, value,
        avg(value) OVER (ORDER BY ts ROWS BETWEEN 60 PRECEDING AND CURRENT ROW) AS mu,
        stddevSamp(value) OVER (ORDER BY ts ROWS BETWEEN 60 PRECEDING AND CURRENT ROW) AS sigma
    FROM metrics WHERE metric = 'cpu'
)
SELECT ts, value
FROM base
WHERE sigma > 0 AND abs(value - mu) / sigma > 3;

-- 当前值与 7 天前同时刻对比
SELECT
    today.ts, today.value AS now_val, past.value AS past_val,
    (today.value - past.value) / past.value * 100 AS change_pct
FROM (SELECT toStartOfInterval(ts, INTERVAL 1 HOUR) AS ts, avg(value) AS value
      FROM metrics WHERE metric = 'requests' AND ts >= today() GROUP BY ts) today
LEFT JOIN (SELECT toStartOfInterval(ts, INTERVAL 1 HOUR) AS ts, avg(value) AS value
           FROM metrics WHERE metric = 'requests'
           AND ts >= today() - INTERVAL 7 DAY AND ts < today() - INTERVAL 6 DAY
           GROUP BY ts) past
ON today.ts = past.ts
ORDER BY change_pct DESC;