Apache Iceberg 데이터 레이크하우스: 테이블 포맷 혁신과 실전 운영 가이드

┌──────────────────────────────────────────────────────────────┐
│                    Iceberg Catalog                            │
│            (현재 메타데이터 파일 위치 포인터)                     │
└──────────────────────┬───────────────────────────────────────┘
                       │
              ┌────────▼────────┐
              │  Metadata File  │  ← JSON 포맷
              │  - 스키마 정보     │
              │  - 파티션 스펙     │
              │  - 스냅샷 목록     │
              │  - 테이블 속성     │
              └────────┬────────┘
                       │
              ┌────────▼────────┐
              │  Manifest List  │  ← Avro 포맷 (스냅샷당 1개)
              │  - 매니페스트 목록  │
              │  - 파티션 요약 통계 │
              └────────┬────────┘
                       │
           ┌───────────┼───────────┐
           │           │           │
    ┌──────▼──────┐ ┌──▼──────┐ ┌─▼──────────┐
    │ Manifest 1  │ │Manifest │ │ Manifest 3 │  ← Avro 포맷
    │ - 파일 경로   │ │   2     │ │ - 파일 경로   │
    │ - 파티션 값   │ │         │ │ - 통계 정보   │
    │ - 컬럼 통계   │ │         │ │ - min/max   │
    └──────┬──────┘ └────┬────┘ └─────┬──────┘
           │             │            │
    ┌──────▼──────┐ ┌────▼────┐ ┌────▼────────┐
    │ Data Files  │ │  Data   │ │ Data Files  │  ← Parquet/ORC/Avro
    │ (Parquet)   │ │ Files   │ │ (Parquet)   │
    └─────────────┘ └─────────┘ └─────────────┘

-- 특정 타임스탬프 기준 조회
SELECT * FROM prod.analytics.user_events
  TIMESTAMP AS OF '2026-03-13 10:00:00';

-- 특정 스냅샷 ID 기준 조회
SELECT * FROM prod.analytics.user_events
  VERSION AS OF 4023948520937184;

-- 태그 또는 브랜치 기준 조회
SELECT * FROM prod.analytics.user_events
  VERSION AS OF 'audit-2026-q1';

-- 스냅샷 히스토리 확인
SELECT * FROM prod.analytics.user_events.snapshots;

-- 스냅샷 간 변경된 데이터 확인 (Incremental Read)
SELECT * FROM prod.analytics.user_events
  FOR SYSTEM_TIME AS OF '2026-03-13 10:00:00';

-- 잘못된 변경 롤백
CALL prod.system.rollback_to_snapshot('analytics.user_events', 4023948520937184);

-- 컬럼 추가
ALTER TABLE prod.analytics.user_events
  ADD COLUMNS (
    device_type STRING COMMENT '접속 디바이스 유형',
    app_version STRING COMMENT '앱 버전'
  );

-- 컬럼 이름 변경 (기존 데이터 파일 재작성 불필요)
ALTER TABLE prod.analytics.user_events
  RENAME COLUMN device_type TO device_category;

-- 컬럼 타입 변경 (안전한 프로모션만 허용)
-- int -> bigint, float -> double 등
ALTER TABLE prod.analytics.user_events
  ALTER COLUMN event_count TYPE BIGINT;

-- 컬럼 삭제
ALTER TABLE prod.analytics.user_events
  DROP COLUMN app_version;

-- 컬럼 순서 변경
ALTER TABLE prod.analytics.user_events
  ALTER COLUMN device_category AFTER event_name;

-- 중첩 구조체 필드 추가
ALTER TABLE prod.analytics.user_events
  ADD COLUMNS (
    metadata.source STRING,
    metadata.campaign_id STRING
  );

-- 테이블 생성 시 파티션 변환 함수 지정
CREATE TABLE prod.analytics.user_events (
    event_id    BIGINT,
    user_id     BIGINT,
    event_name  STRING,
    event_time  TIMESTAMP,
    country     STRING,
    amount      DECIMAL(10, 2)
) USING iceberg
PARTITIONED BY (
    days(event_time),   -- 날짜 기반 자동 파티셔닝
    bucket(16, user_id) -- user_id 해시 기반 버킷팅
);

-- 사용자는 파티션을 의식하지 않고 쿼리 작성
-- Iceberg가 자동으로 파티션 프루닝 수행
SELECT * FROM prod.analytics.user_events
WHERE event_time >= '2026-03-01'
  AND event_time < '2026-03-14'
  AND user_id = 12345;

-- 기존: 일 단위 파티셔닝
-- 트래픽 증가로 시간 단위로 변경 필요

-- 기존 파티션 필드 교체
ALTER TABLE prod.analytics.user_events
  REPLACE PARTITION FIELD days(event_time) WITH hours(event_time);

-- 또는 단계별 변경
ALTER TABLE prod.analytics.user_events
  DROP PARTITION FIELD days(event_time);

ALTER TABLE prod.analytics.user_events
  ADD PARTITION FIELD hours(event_time);

-- Spark SQL에서 Iceberg 카탈로그 설정
-- spark-defaults.conf 또는 SparkSession 생성 시 설정
-- spark.sql.catalog.prod = org.apache.iceberg.spark.SparkCatalog
-- spark.sql.catalog.prod.type = hive
-- spark.sql.catalog.prod.uri = thrift://metastore-host:9083

-- 데이터베이스 생성
CREATE DATABASE IF NOT EXISTS prod.analytics;

-- Iceberg 테이블 생성 (전체 옵션)
CREATE TABLE prod.analytics.orders (
    order_id      BIGINT,
    customer_id   BIGINT,
    product_name  STRING,
    quantity       INT,
    unit_price    DECIMAL(10, 2),
    order_status  STRING,
    created_at    TIMESTAMP,
    updated_at    TIMESTAMP
) USING iceberg
PARTITIONED BY (days(created_at), bucket(8, customer_id))
TBLPROPERTIES (
    'write.format.default'          = 'parquet',
    'write.target-file-size-bytes'  = '536870912',
    'write.parquet.compression-codec' = 'zstd',
    'read.split.target-size'        = '134217728',
    'history.expire.max-snapshot-age-ms' = '432000000',
    'write.metadata.delete-after-commit.enabled' = 'true',
    'write.metadata.previous-versions-max' = '100'
);

-- 데이터 삽입
INSERT INTO prod.analytics.orders VALUES
    (1, 1001, 'Laptop', 1, 1299.99, 'completed', TIMESTAMP '2026-03-14 09:00:00', TIMESTAMP '2026-03-14 09:00:00'),
    (2, 1002, 'Mouse',  2, 29.99,   'pending',   TIMESTAMP '2026-03-14 09:15:00', TIMESTAMP '2026-03-14 09:15:00');

-- MERGE INTO (Upsert)
MERGE INTO prod.analytics.orders t
USING staging.new_orders s
ON t.order_id = s.order_id
WHEN MATCHED THEN
    UPDATE SET order_status = s.order_status, updated_at = s.updated_at
WHEN NOT MATCHED THEN
    INSERT *;

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, lit

# SparkSession 생성 (Iceberg 카탈로그 설정 포함)
spark = SparkSession.builder \
    .appName("IcebergExample") \
    .config("spark.sql.catalog.prod", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.prod.type", "hive") \
    .config("spark.sql.catalog.prod.uri", "thrift://metastore-host:9083") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.7.1") \
    .getOrCreate()

# DataFrame API로 Iceberg 테이블 읽기
df = spark.read.format("iceberg").load("prod.analytics.orders")
df.show()

# 타임 트래블 (특정 스냅샷 ID로 읽기)
df_snapshot = spark.read \
    .option("snapshot-id", 4023948520937184) \
    .format("iceberg") \
    .load("prod.analytics.orders")

# 타임 트래블 (특정 타임스탬프로 읽기)
df_time = spark.read \
    .option("as-of-timestamp", "1710400800000") \
    .format("iceberg") \
    .load("prod.analytics.orders")

# Incremental Read (두 스냅샷 사이의 변경 데이터)
df_incremental = spark.read \
    .option("start-snapshot-id", 1000) \
    .option("end-snapshot-id", 2000) \
    .format("iceberg") \
    .load("prod.analytics.orders")

# DataFrame API로 쓰기
new_orders_df = spark.createDataFrame([
    (3, 1003, "Keyboard", 1, 79.99, "completed", "2026-03-14 10:00:00", "2026-03-14 10:00:00"),
], ["order_id", "customer_id", "product_name", "quantity", "unit_price", "order_status", "created_at", "updated_at"])

new_orders_df.writeTo("prod.analytics.orders").append()

# 메타데이터 조회
snapshots_df = spark.read.format("iceberg").load("prod.analytics.orders.snapshots")
snapshots_df.show(truncate=False)

history_df = spark.read.format("iceberg").load("prod.analytics.orders.history")
history_df.show(truncate=False)

# 매니페스트 파일 정보 조회
manifests_df = spark.read.format("iceberg").load("prod.analytics.orders.manifests")
manifests_df.show(truncate=False)

# 파일 레벨 통계 조회
files_df = spark.read.format("iceberg").load("prod.analytics.orders.files")
files_df.select("file_path", "file_size_in_bytes", "record_count").show(truncate=False)

-- Trino catalog 설정 (etc/catalog/iceberg.properties)
-- connector.name=iceberg
-- hive.metastore.uri=thrift://metastore-host:9083
-- iceberg.file-format=PARQUET
-- iceberg.compression-codec=ZSTD

-- Trino에서 Iceberg 테이블 쿼리
SELECT order_status, COUNT(*) as order_count, SUM(unit_price * quantity) as total_revenue
FROM iceberg.analytics.orders
WHERE created_at >= TIMESTAMP '2026-03-01 00:00:00'
GROUP BY order_status;

-- Trino에서 타임 트래블
SELECT * FROM iceberg.analytics.orders
FOR TIMESTAMP AS OF TIMESTAMP '2026-03-13 00:00:00';

-- 테이블 메타데이터 확인
SELECT * FROM iceberg.analytics."orders$snapshots";
SELECT * FROM iceberg.analytics."orders$manifests";
SELECT * FROM iceberg.analytics."orders$files";

-- Bin-Pack 컴팩션 (기본 전략, 빠른 실행)
CALL prod.system.rewrite_data_files(
    table => 'analytics.orders',
    strategy => 'binpack',
    options => map(
        'target-file-size-bytes', '536870912',
        'min-file-size-bytes',    '402653184',
        'max-file-size-bytes',    '671088640',
        'max-file-group-size-bytes', '107374182400',
        'partial-progress.enabled', 'true',
        'partial-progress.max-commits', '10'
    )
);

-- Sort 컴팩션 (읽기 성능 최적화, 더 오래 걸림)
CALL prod.system.rewrite_data_files(
    table => 'analytics.orders',
    strategy => 'sort',
    sort_order => 'customer_id ASC NULLS LAST, created_at DESC',
    options => map(
        'target-file-size-bytes', '536870912'
    )
);

-- Z-Order 컴팩션 (다차원 쿼리 최적화)
CALL prod.system.rewrite_data_files(
    table => 'analytics.orders',
    strategy => 'sort',
    sort_order => 'zorder(customer_id, product_name)',
    options => map(
        'target-file-size-bytes', '536870912'
    )
);

-- 특정 파티션만 컴팩션
CALL prod.system.rewrite_data_files(
    table => 'analytics.orders',
    where => 'created_at >= TIMESTAMP ''2026-03-01'' AND created_at < TIMESTAMP ''2026-03-14'''
);

-- 5일보다 오래된 스냅샷 만료
CALL prod.system.expire_snapshots(
    table => 'analytics.orders',
    older_than => TIMESTAMP '2026-03-09 00:00:00',
    retain_last => 5,
    max_concurrent_deletes => 16
);

-- 고아 파일 제거 (최소 3일 이상 된 파일만)
CALL prod.system.remove_orphan_files(
    table => 'analytics.orders',
    older_than => TIMESTAMP '2026-03-11 00:00:00',
    dry_run => true  -- 먼저 dry_run으로 확인
);

CALL prod.system.rewrite_manifests('analytics.orders');