Auto Loader 스키마 실전 패턴

수요일 14:00 — 모바일 앱 팀이 주문 API에 새 필드 추가
  - 기존: {"order_id": 123, "amount": 5000, "user_id": "abc"}
  - 변경: {"order_id": 123, "amount": 5000, "user_id": "abc",
            "coupon_code": "SUMMER25", "discount_rate": 0.15}
  - 앱 팀은 "하위 호환이니까 데이터팀에 안 알려도 되겠지" 라고 판단

수요일 14:05 — Auto Loader가 새 파일을 읽기 시작
  - schemaEvolutionMode = "none" 이었다면 → 새 필드가 조용히 사라짐 (데이터 유실!)
  - schemaEvolutionMode = "failOnNewColumns" 이었다면 → 스트림 즉시 중단
  - schemaEvolutionMode = "addNewColumns" 이었다면 → 스트림 중단 후 재시작 시 자동 반영

수요일 14:10 — 더 심각한 상황: 타입 변경
  - 앱 팀이 amount를 정수에서 문자열로 변경 (쿠폰 적용 시 "5000원 - 750원 할인")
  - 기존: {"amount": 5000}
  - 변경: {"amount": "5000원 - 750원 할인"}
  - → inferColumnTypes로 BIGINT로 추론된 컬럼에 STRING이 들어옴
  - → 파싱 실패! 파이프라인 장애!

목요일 새벽 03:00 — 하루치 데이터가 없다는 알림
  - 마케팅팀의 일간 매출 대시보드가 빈 값을 보여줌
  - 당직 엔지니어에게 전화가 옴 📞

-- rescued_data에 데이터가 쌓이기 시작하면 = 스키마가 변경되었다는 신호
-- 이 쿼리를 DBSQL Alert로 등록하세요

SELECT
    DATE(ingested_at) AS dt,
    COUNT(*) AS total_rows,
    COUNT(_rescued_data) AS rescued_rows,
    ROUND(COUNT(_rescued_data) * 100.0 / COUNT(*), 2) AS rescued_pct
FROM catalog.bronze.orders
WHERE ingested_at >= CURRENT_DATE() - INTERVAL 1 DAY
GROUP BY 1;

-- rescued_pct가 0%에서 갑자기 올라가면 → 소스 스키마가 변경된 것!
-- 알림 조건: rescued_pct > 0

-- rescued_data에 어떤 필드가 들어오는지 분석
SELECT
    json_key,
    COUNT(*) AS occurrences,
    MIN(ingested_at) AS first_seen,
    MAX(ingested_at) AS last_seen
FROM (
    SELECT
        explode(map_keys(from_json(_rescued_data,
            'MAP<STRING, STRING>'))) AS json_key,
        _ingested_at AS ingested_at
    FROM catalog.bronze.orders
    WHERE _rescued_data IS NOT NULL
)
GROUP BY json_key
ORDER BY occurrences DESC;

-- 결과 예시:
-- | json_key      | occurrences | first_seen          | last_seen           |
-- |---------------|-------------|---------------------|---------------------|
-- | coupon_code   | 15,230      | 2025-03-15 14:05:00 | 2025-03-15 23:59:00 |
-- | discount_rate | 15,230      | 2025-03-15 14:05:00 | 2025-03-15 23:59:00 |

-- Silver 레이어에서 rescued_data를 파싱하여 정식 컬럼으로 추출
CREATE OR REFRESH STREAMING TABLE silver_orders AS
SELECT
    *,
    -- rescued_data에서 새 필드를 추출
    COALESCE(
        coupon_code,
        get_json_object(_rescued_data, '$.coupon_code')
    ) AS coupon_code_final,
    COALESCE(
        discount_rate,
        CAST(get_json_object(_rescued_data, '$.discount_rate') AS DOUBLE)
    ) AS discount_rate_final
FROM STREAM(catalog.bronze.orders);

# ❌ 위험한 설정: schemaHints 없이 모든 것을 추론에 맡김
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .load("s3://bucket/data/")
)
# → 첫 번째 파일에 amount가 5000(정수)이면 BIGINT로 추론
# → 나중에 amount가 5000.50(소수)이면 타입 불일치로 rescued_data로 빠짐

# ✅ 안전한 설정: 핵심 컬럼에 schemaHints 적용
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaHints", """
        order_id BIGINT,
        amount DECIMAL(18,4),
        user_id STRING,
        created_at TIMESTAMP,
        items ARRAY<STRUCT<product_id:STRING, qty:INT, price:DECIMAL(12,2)>>
    """)
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/")
    .option("rescuedDataColumn", "_rescued_data")
    .load("s3://bucket/data/")
)

# 20년 경험에서 나온 "황금 조합"
df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    # 1. 타입 추론 활성화 (편의성)
    .option("cloudFiles.inferColumnTypes", "true")
    # 2. 핵심 컬럼은 schemaHints로 고정 (안전성)
    .option("cloudFiles.schemaHints",
            "order_id STRING, amount DECIMAL(18,4), created_at TIMESTAMP")
    # 3. 새 컬럼은 자동 추가 (유연성)
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    # 4. 파싱 실패 데이터는 rescued_data에 보존 (데이터 무결성)
    .option("rescuedDataColumn", "_rescued_data")
    # 5. 스키마 위치 고유하게 설정
    .option("cloudFiles.schemaLocation", "s3://bucket/schema/orders_v1/")
    .load("s3://bucket/raw/orders/")
)