통신 단말의 데이터를 사람, 서비스, AI 에이전트가 함께 쓰는 OSS 레이크

네 단계는 데이터베이스 oss_lake의 ue_trace_raw, ue_trace, cell_kpi_1m, cell_rank_now 테이블입니다. 앞의 둘은 Materialized View로, 마지막은 1분마다 도는 Refreshable Materialized View로 이어집니다. 유즈케이스는 04-usecases.sql의 뷰입니다.

측정 한 건의 전과 후

412행이 접힌 1분 요약 한 행

bucket	2026-10-18 09:17:00
cell	617853093946327039
users_st	uniqState(…) → 146
samples_n	412
rsrp_n	364
rsrp_sum	-28464
poor_n	0

관제 지도가 읽는 20행

만들며 알게 된 함정. 집계 상태 타입은 원본 컬럼의 Nullable 여부와 맞아야 합니다. 맞지 않는 MV는 CREATE에서 거부되지만, 나중에 ALTER로 원본 컬럼을 Nullable로 바꾸면 그 뒤 INSERT가 모두 실패하고 원본만 계속 쌓입니다(README에 재현 기록).

정의

-- ---------------------------------------------------------------- bronze
-- Exactly what the probes send: H3 as a string, the IMSI as the vendor's
-- ciphertext (44-char base64), every measurement Nullable, CQI as text.
CREATE TABLE oss_lake.ue_trace_raw
(
    event_ts   DateTime64(3),
    h3_res09   Nullable(String),
    imsi       Nullable(String),
    rsrp       Nullable(Float64),   -- dBm
    rsrq       Nullable(Float64),   -- dB
    sinr       Nullable(Float64),   -- dB
    cqi        Nullable(String),
    pci        Nullable(UInt16),
    probe      LowCardinality(String),
    ingest_ts  DateTime64(3) DEFAULT now64(3)
)
ENGINE = MergeTree
PARTITION BY toDate(event_ts)
ORDER BY event_ts
TTL toDateTime(event_ts) + INTERVAL 3 DAY;

-- 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
FROM oss_lake.ue_trace_raw
WHERE h3_res09 IS NOT NULL AND imsi IS NOT NULL;

-- ---------------------------------------------------------------- gold
-- One row per (minute, cell). Users are a uniq state so any time range can be merged;
-- the rest are plain sums. Column names differ from the aliases queries use
-- (users_st vs users): an alias equal to the column inside its own aggregate is
-- ILLEGAL_AGGREGATION.
CREATE TABLE oss_lake.cell_kpi_1m
(
    bucket     DateTime,
    cell       UInt64,
    users_st   AggregateFunction(uniq, UInt64),
    samples_n  SimpleAggregateFunction(sum, UInt64),
    rsrp_n     SimpleAggregateFunction(sum, UInt64),   -- samples that carried rsrp
    rsrp_sum   SimpleAggregateFunction(sum, Float64),
    poor_n     SimpleAggregateFunction(sum, UInt64)    -- rsrp < -100 dBm
)
ENGINE = AggregatingMergeTree
ORDER BY (bucket, cell);

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;

-- 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,089
3. 1분 요약 골드2,431,516
4. 바로 쓰는 답 서빙20

• 받은 그대로 1,004.6 ms · 10,807,707행 읽음
• 다듬기 289.9 ms · 10,742,429행 읽음
• 1분 요약 249.8 ms · 1,219,101행 읽음
• 바로 쓰는 답 1.4 ms · 20행 읽음

측정 2026-10-18 09:19 UTC, ClickHouse 26.9.13.15, Apple M4 Pro (macOS arm64), 컨테이너 CPU 12개 · 메모리 7.75 GiB. 워밍업 1회 뒤 5회 실행의 중앙값, 서버 측 경과 시간, 쿼리 캐시 끔. 참고용입니다.