通信端末のデータを人、サービス、AI エージェントが一緒に使う OSS レイク

四つの段階はデータベース oss_lake の ue_trace_raw、ue_trace、cell_kpi_1m、cell_rank_now テーブルです。最初の二つはマテリアライズドビューで、最後は毎分動くリフレッシュ可能なマテリアライズドビューでつながります。ユースケースは 04-usecases.sql のビューです。

測定 1 件の前と後

441 行が畳まれた 1 分要約の 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

監視マップが読む 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;

どの段階に聞いても同じ答え

同じ監視の質問を四つの段階それぞれに投げると、同じ 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)、コンテナの CPU 12 個 · メモリ 7.75 GiB。ウォームアップ 1 回の後 5 回実行の中央値、サーバー側の経過時間、クエリキャッシュ無効。参考値です。

作りながら分かった落とし穴。 集計状態の型は、元の列が Nullable かどうかと一致している必要があります。一致しない MV は CREATE で拒否されますが、後から ALTER で元の列を Nullable にすると、その後の INSERT はすべて失敗し、元のテーブルだけが増え続けます(README に再現記録)。