让人、服务和 AI 智能体共用所有通信终端数据的 OSS 数据湖

四个阶段是数据库 oss_lake 中的 ue_trace_raw、ue_trace、cell_kpi_1m 和 cell_rank_now 表。前两段由物化视图连接，最后一段由每分钟运行的可刷新物化视图连接。用例是 04-usecases.sql 中的视图。

一条测量的前后对比

441 行折成的 1 分钟汇总行

bucket	2026-10-18 10:20:00
cell	617853093946327039
users_st	uniqState(…) → 133
samples_n	441
rsrp_n	399
rsrp_sum	-31229
poor_n	0

NOC 地图读取的 20 行

定义

-- ---------------------------------------------------------------- the inventory (CM)
-- Every cell gets a site and a fictional district: 'Central' within 10 rings, otherwise one
-- of six 60-degree sectors around the city centre. Coverage holes are known weak spots.
INSERT INTO oss_lake.cell_inventory
WITH
    [(37.5665, 126.9780), (35.1796, 129.0756), (35.8714, 128.6014)] AS centres,
    ['North-East', 'East', 'South-East', 'South-West', 'West', 'North-West'] AS sectors,
    centres[city + 1] AS c,
    h3ToGeo(h3) AS p,
    (degrees(atan2((p.2 - c.2) * cos(radians(c.1)), p.1 - c.1)) + 360) % 360 AS bearing
SELECT
    h3                                                            AS cell,
    concat(['A', 'B', 'C'][city + 1], '-', leftPad(toString(intDiv(idx, 7)), 4, '0')) AS site_id,
    ['City A', 'City B', 'City C'][city + 1]                      AS region1,
    concat(['A', 'B', 'C'][city + 1], ' ',
           if(ring <= 10, 'Central', sectors[1 + toUInt8(floor(bearing / 60)) % 6])) AS region2,
    is_hole                                                       AS known_weak
FROM oss_lake.grid;

-- Rows without a cell or a subscriber cannot be placed or counted: silver drops them.
CREATE MATERIALIZED VIEW oss_lake.ue_trace_mv TO oss_lake.ue_trace AS
SELECT
    event_ts,
    stringToH3(assumeNotNull(h3_res09))       AS h3,
    cityHash64(assumeNotNull(imsi))           AS imsi_hash,
    CAST(rsrp AS Nullable(Float32))           AS rsrp,
    toFloat32(ifNull(rsrq, 0))                AS rsrq,
    toFloat32(ifNull(sinr, 0))                AS sinr,
    toUInt8OrZero(ifNull(cqi, ''))            AS cqi,
    -- a measurement gets a place name from the inventory
    dictGet('oss_lake.cell_dict', 'region1', h3) AS region1,
    dictGet('oss_lake.cell_dict', 'region2', h3) AS region2
FROM oss_lake.ue_trace_raw
WHERE h3_res09 IS NOT NULL AND imsi IS NOT NULL;

CREATE MATERIALIZED VIEW oss_lake.cell_kpi_1m_mv TO oss_lake.cell_kpi_1m AS
SELECT
    toStartOfMinute(event_ts)  AS bucket,
    h3                         AS cell,
    uniqState(imsi_hash)       AS users_st,
    count()                    AS samples_n,
    count(rsrp)                AS rsrp_n,
    sum(ifNull(rsrp, 0))       AS rsrp_sum,
    countIf(rsrp < -100)       AS poor_n
FROM oss_lake.ue_trace
GROUP BY bucket, cell;

CREATE MATERIALIZED VIEW oss_lake.region_kpi_1m_mv TO oss_lake.region_kpi_1m AS
SELECT
    toStartOfMinute(event_ts)  AS bucket,
    region1, region2,
    uniqState(imsi_hash)       AS users_st,
    count()                    AS samples_n,
    count(rsrp)                AS rsrp_n,
    sum(ifNull(rsrp, 0))       AS rsrp_sum,
    countIf(rsrp < -100)       AS poor_n
FROM oss_lake.ue_trace
GROUP BY bucket, region1, region2;

[[sql:gold10]]

-- No APPEND: every refresh replaces the 20 rows.
CREATE MATERIALIZED VIEW oss_lake.cell_rank_rmv
REFRESH EVERY 1 MINUTE TO oss_lake.cell_rank_now AS
WITH (SELECT max(anchor_ts) FROM oss_lake.params) AS t
SELECT
    lr.1 AS list, lr.2 AS rnk, cell, users, samples, avg_rsrp, poor_pct, t AS window_end
FROM
(
    SELECT
        cell,
        uniqMerge(users_st)                              AS users,
        sum(samples_n)                                   AS samples,
        round(sum(rsrp_sum) / sum(rsrp_n), 1)            AS avg_rsrp,
        round(100 * sum(poor_n) / sum(rsrp_n), 1)        AS poor_pct,
        sum(rsrp_n) >= 30                                AS eligible,
        row_number() OVER (ORDER BY users DESC, cell)    AS r_users,
        row_number() OVER (PARTITION BY eligible ORDER BY sum(rsrp_sum) / sum(rsrp_n), cell) AS r_signal
    FROM oss_lake.cell_kpi_1m
    WHERE bucket >= t - INTERVAL 1 HOUR AND bucket < t
    GROUP BY cell
)
ARRAY JOIN [('users', r_users), ('signal', if(eligible, r_signal, 999))] AS lr
WHERE lr.2 <= 10;

-- ---------------------------------------------------------------- 3 · alerts
-- Cells whose signal in the latest minute is at least 12 dB below their own last hour.
-- Subscribers (an on-call app, a chat bot) pick these up every minute; see 4 and
-- subscribe-kafka.sql for how they are delivered.
CREATE OR REPLACE VIEW oss_lake.uc_alert_cells AS
WITH (SELECT max(anchor_ts) FROM oss_lake.params) AS t
SELECT
    cell, city, area, minute, rsrp_now, rsrp_hour,
    round(rsrp_hour - rsrp_now, 1)                           AS drop_db,
    samples_now
FROM
(
    SELECT
        k.cell                                               AS cell,
        any(g.city)                                          AS city,
        h3ToParent(k.cell, 7)                                AS area,
        t - INTERVAL 1 MINUTE                                AS minute,
        sumIf(k.rsrp_sum, k.bucket = minute) / sumIf(k.rsrp_n, k.bucket = minute) AS rsrp_now,
        sum(k.rsrp_sum) / sum(k.rsrp_n)                      AS rsrp_hour,
        sumIf(k.rsrp_n, k.bucket = minute)                   AS samples_now
    FROM oss_lake.cell_kpi_1m AS k
    INNER JOIN oss_lake.grid AS g ON g.h3 = k.cell
    WHERE k.bucket >= t - INTERVAL 1 HOUR AND k.bucket < t
    GROUP BY cell
)
WHERE samples_now >= 3 AND rsrp_hour - rsrp_now >= 12;

问哪个阶段都得到同样的答案

把同一个 NOC 问题分别发给四个阶段，返回的 20 行完全相同（由 03-queries.sql 检查）。每个阶段的行数和读取量都在减少。

1. 原样 青铜层21,600,000
2. 整理 白银层21,470,680
3. 1 分钟汇总 黄金层2,431,524
4. 现成的答案 服务层20

• 原样 710.6 ms · 读取 10,807,707 行
• 整理 229.7 ms · 读取 10,742,997 行
• 1 分钟汇总 185 ms · 读取 1,219,109 行
• 现成的答案 1.2 ms · 读取 20 行

测量于 2026-10-18 10:22 UTC，ClickHouse 26.9.13.15，Apple M4 Pro (macOS arm64)，容器 12 个 CPU · 内存 7.75 GiB。预热 1 次后 5 次运行的中位数，服务器端耗时，查询缓存关闭。仅供参考。

构建时发现的一个陷阱。 聚合状态的类型必须与源列是否 Nullable 一致。不一致的 MV 在 CREATE 时就被拒绝；但如果之后用 ALTER 把源列改成 Nullable，此后的 INSERT 全部失败，而源表照样增长（README 中有复现记录）。