StarRocks 도입기: 실시간 OLAP으로 데이터 파이프라인을 뜯어고친 이야기

Service → Kafka → Iceberg → S3 → Trino → Airflow(5분) → Dashboard

Service → Kafka → StarRocks → Dashboard (서브초 레이턴시)

┌─────────────────────────────┐
│  어떤 데이터를 저장하는가?    │
└──────────────┬──────────────┘
               │
       ┌───────▼────────┐
       │  UPDATE 필요?   │
       └───────┬────────┘
               │
      ┌────────┴────────┐
      │                 │
    [아니오]            [예]
      │                 │
┌─────▼─────┐    ┌─────▼──────┐
│  집계 필요? │    │ Primary Key │
└─────┬─────┘    │ (빈번한     │
      │          │  UPDATE)    │
  [아니오]  [예]  └────────────┘
      │      │
┌─────▼───┐ ┌▼──────────┐
│Duplicate│ │Aggregate   │
│(원본 저장)│ │(자동 집계) ★│
└─────────┘ └────────────┘

CREATE TABLE order_events (
    event_id BIGINT,
    event_time DATETIME,
    order_id VARCHAR(50),
    user_id BIGINT,
    event_type VARCHAR(20),
    amount DECIMAL(10, 2)
)
DUPLICATE KEY(event_id, event_time)
PARTITION BY date_trunc('day', event_time)
DISTRIBUTED BY HASH(event_id) BUCKETS 10;

CREATE TABLE order_stats (
    stat_time DATETIME NOT NULL COMMENT '5분 간격',
    region VARCHAR(20) NOT NULL,
    delivery_type VARCHAR(20) NOT NULL,
    -- 집계 컬럼: 수집 시 자동으로 집계 함수 적용
    order_count BIGINT SUM DEFAULT "0",
    total_amount DECIMAL(15, 2) SUM DEFAULT "0",
    user_bitmap BITMAP BITMAP_UNION,
    max_amount DECIMAL(10, 2) MAX
)
AGGREGATE KEY(stat_time, region, delivery_type)
PARTITION BY date_trunc('day', stat_time)
DISTRIBUTED BY HASH(stat_time) BUCKETS 10;

CREATE TABLE orders (
    order_id VARCHAR(50) NOT NULL,
    status VARCHAR(20),
    amount DECIMAL(10, 2),
    updated_at DATETIME
)
PRIMARY KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 10
PROPERTIES (
    "enable_persistent_index" = "true"
);

CREATE ROUTINE LOAD order_load ON orders
COLUMNS(
    order_id,
    user_id,
    timestamp_ms,
    amount,
    order_date = FROM_UNIXTIME(timestamp_ms / 1000)
)
PROPERTIES (
    "format" = "json",
    "jsonpaths" = "[\"$.orderId\",\"$.userId\",\"$.timestamp\",\"$.amount\"]"
)
FROM KAFKA (
    "kafka_broker_list" = "kafka-broker:9092",
    "kafka_topic" = "orders"
);

CREATE ROUTINE LOAD order_stats_load ON order_stats
COLUMNS(
    timestamp_ms,
    region,
    amount,
    user_id,
    -- 5분 간격으로 라운딩
    stat_time = FROM_UNIXTIME(FLOOR(timestamp_ms / 1000 / 300) * 300),
    order_count = 1,
    total_amount = amount,
    user_bitmap = BITMAP_HASH(user_id)
)
WHERE amount > 0
PROPERTIES ("format" = "json")
FROM KAFKA (
    "kafka_broker_list" = "kafka-broker:9092",
    "kafka_topic" = "orders"
);

# CSV 파일 로딩
curl --location-trusted \
  -u user:password \
  -H "label:load_$(date +%Y%m%d%H%M%S)" \
  -H "column_separator:," \
  -T data.csv \
  http://starrocks-fe:8030/api/mydb/mytable/_stream_load

# be.conf
pipeline_scan_thread_pool_thread_num = 32   # 기본값: 24
pipeline_exec_thread_pool_thread_num = 32   # 기본값: 24

-- ✅ 올바른 사용: 파티션 프루닝 동작
WHERE event_time >= NOW() - INTERVAL 3 DAY

-- ❌ 잘못된 사용: 파티션 프루닝 불가
WHERE DATE(event_time) >= CURRENT_DATE - 3

PROPERTIES (
    "partition_live_number" = "3"  -- 최근 3개 파티션만 유지
)

-- 상태 확인
SHOW MATERIALIZED VIEWS;

-- 강제 동기 리프레시
REFRESH MATERIALIZED VIEW db.mv_name WITH SYNC MODE;

-- 비활성화된 MV 재활성화
ALTER MATERIALIZED VIEW db.mv_name ACTIVE;

-- 상태 확인
SHOW ROUTINE LOAD FOR db.load_job;

-- 재개
RESUME ROUTINE LOAD FOR db.load_job;

-- 1. 현재 노드 확인
SHOW PROC '/backends';

-- 2. 디커미션 시작
ALTER SYSTEM DECOMMISSION BACKEND "<BE_IP>:<HEARTBEAT_PORT>";

-- 3. TabletNum이 0이 될 때까지 대기 후 제거
ALTER SYSTEM DROP BACKEND "<BE_IP>:<HEARTBEAT_PORT>";