스키마 진화(Schema Evolution)와 스키마 강제(Schema Enforcement)

-- 테이블 정의
CREATE TABLE catalog.schema.products (
    product_id BIGINT,
    name STRING,
    price DECIMAL(10, 2)
);

-- ✅ 정상: 스키마가 일치
INSERT INTO catalog.schema.products VALUES (1, '노트북', 1500000.00);

-- ❌ 에러: 'category' 컬럼이 테이블에 없음
INSERT INTO catalog.schema.products (product_id, name, price, category)
VALUES (2, '키보드', 89000.00, '전자제품');
-- AnalysisException: Cannot write to table because the column 'category' is not found

-- ❌ 에러: price에 문자열 삽입 시도
INSERT INTO catalog.schema.products VALUES (3, '마우스', 'cheap');
-- AnalysisException: Cannot safely cast 'price': string to decimal(10,2)

-- 컬럼 추가
ALTER TABLE catalog.schema.products
ADD COLUMN category STRING;

-- 컬럼 추가 (기본값 포함)
ALTER TABLE catalog.schema.products
ADD COLUMN created_at TIMESTAMP DEFAULT current_timestamp();

-- 컬럼 이름 변경 (Column Mapping 필요)
ALTER TABLE catalog.schema.products
RENAME COLUMN name TO product_name;

-- 컬럼 삭제 (Column Mapping 필요)
ALTER TABLE catalog.schema.products
DROP COLUMN category;

-- 컬럼 타입 변경 (안전한 확장만 가능)
ALTER TABLE catalog.schema.products
ALTER COLUMN product_id TYPE BIGINT;  -- INT → BIGINT은 가능

-- SQL: INSERT 시 mergeSchema 활성화
SET spark.databricks.delta.schema.autoMerge.enabled = true;

INSERT INTO catalog.schema.products VALUES
(4, '태블릿', 800000.00, '전자제품', current_timestamp());
-- 'category' 컬럼이 자동으로 추가됩니다

# PySpark: DataFrame 쓰기 시 mergeSchema 옵션
df_new = spark.createDataFrame([
    (5, "스피커", 250000.00, "전자제품", "2025-03-20")
], ["product_id", "name", "price", "category", "launch_date"])

df_new.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("catalog.schema.products")

-- MERGE에서도 스키마 진화가 가능합니다
SET spark.databricks.delta.schema.autoMerge.enabled = true;

MERGE INTO catalog.schema.products AS target
USING new_products AS source
ON target.product_id = source.product_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
-- source에 새 컬럼이 있으면 자동으로 추가됩니다

-- Column Mapping 활성화
ALTER TABLE catalog.schema.products
SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name');

-- Column Mapping이 활성화된 테이블에서

-- 컬럼 이름 변경
ALTER TABLE catalog.schema.products
RENAME COLUMN name TO product_name;

-- 컬럼 삭제
ALTER TABLE catalog.schema.products
DROP COLUMN old_unused_column;

-- 여러 컬럼 동시 삭제
ALTER TABLE catalog.schema.products
DROP COLUMNS (temp_col1, temp_col2);

# Auto Loader에서 스키마 진화 활성화
(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/schema")
    # 스키마 진화 모드 설정
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("s3://my-bucket/raw-data/")
    .writeStream
    .option("checkpointLocation", "/checkpoints/stream")
    .option("mergeSchema", "true")
    .table("catalog.schema.bronze_events")
)

| Column Mapping 모드 | 논리적 이름 | 물리적 이름 | 설명 |
|--------------------|-----------|-----------|------|
| `none` (기본, 비활성) | customer_name | customer_name | 동일 |
| `name` (활성화) | customer_name | col-a1b2c3d4-e5f6 | UUID 기반 물리적 이름 |

// Delta Log의 metadata action 내 column mapping 정보
{
  "metaData": {
    "schemaString": "{\"fields\":[{\"name\":\"customer_name\",\"metadata\":{\"delta.columnMapping.id\":1,\"delta.columnMapping.physicalName\":\"col-a1b2c3d4-e5f6\"}}]}",
    "configuration": {
      "delta.columnMapping.mode": "name",
      "delta.columnMapping.maxColumnId": "5"
    }
  }
}

-- Column Mapping = name 일 때
ALTER TABLE products RENAME COLUMN customer_name TO client_name;

-- 실제 일어나는 일:
-- 1. Delta Log에 새 metadata action 기록
-- 2. 논리적 이름: customer_name → client_name (변경)
-- 3. 물리적 이름: col-a1b2c3d4 (변경 없음!)
-- 4. 기존 Parquet 파일: 수정 없음 (제로 카피 연산)
-- → 수 TB 테이블이라도 이름 변경이 밀리초 단위로 완료됩니다

# Column Mapping 변경 후 스트리밍 재시작 절차
# 1. 스트리밍 쿼리 중지
streaming_query.stop()

# 2. 체크포인트 초기화 (또는 삭제)
dbutils.fs.rm("/checkpoints/my_stream", recurse=True)

# 3. 스트리밍 쿼리 재시작 (새 체크포인트 경로)
(spark.readStream.table("source_table")
    .writeStream
    .option("checkpointLocation", "/checkpoints/my_stream_v2")  # 새 경로
    .table("target_table"))

# 100개 이상의 새 컬럼이 있는 DataFrame을 기존 테이블에 병합
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# 새 스키마로 DataFrame 생성 (기존 컬럼 + 새 100개 컬럼)
new_columns = [StructField(f"sensor_{i:03d}", DoubleType(), True) for i in range(100)]
# ... DataFrame 준비 ...

# mergeSchema로 한 번에 추가
df_new_data.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("catalog.schema.sensor_readings")

# 결과: 기존 테이블에 100개의 새 컬럼이 자동 추가됩니다
# 기존 행의 새 컬럼 값은 NULL로 채워집니다

[소스 파일]                  [Auto Loader]              [Delta 테이블]
새 JSON 파일 도착        →   스키마 감지               →  스키마 진화
(새 필드 "loyalty_tier"      (addNewColumns 모드)         (mergeSchema=true)
 포함)                       새 컬럼 발견!                loyalty_tier 컬럼 추가
                             스키마 파일 업데이트          기존 행은 NULL
                             스트림 재시작 (자동)

# 프로덕션에서 안전한 스키마 진화 패턴
# 1단계: rescue 모드로 안전하게 수집
(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/schema")
    .option("cloudFiles.schemaEvolutionMode", "rescue")  # 새 컬럼은 _rescued_data에 저장
    .load("s3://bucket/raw/")
    .writeStream
    .option("checkpointLocation", "/checkpoints/bronze")
    .table("catalog.schema.bronze_events"))

# 2단계: _rescued_data 분석하여 새 컬럼 확인
rescued = spark.sql("""
    SELECT _rescued_data, COUNT(*) as cnt
    FROM catalog.schema.bronze_events
    WHERE _rescued_data IS NOT NULL
    GROUP BY _rescued_data
    ORDER BY cnt DESC
""")

# 3단계: 확인 후 DDL로 명시적으로 스키마 변경
# ALTER TABLE catalog.schema.bronze_events ADD COLUMN loyalty_tier STRING;

# 4단계: 스트림 재시작 (체크포인트 유지, 새 컬럼 포함)

# SDP 파이프라인에서 Auto Loader + 스키마 진화
import dlt

@dlt.table(
    comment="Bronze: 원본 이벤트 데이터"
)
def bronze_events():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("cloudFiles.schemaLocation", "/checkpoints/schema")
        .load("s3://bucket/raw/events/")
    )

@dlt.table(
    comment="Silver: 정제된 이벤트"
)
@dlt.expect_or_drop("valid_event_type", "event_type IS NOT NULL")
def silver_events():
    return dlt.read_stream("bronze_events").select(
        "event_id",
        "event_type",
        "user_id",
        "timestamp"
        # 명시적 컬럼 선택 → Bronze에 새 컬럼이 추가되어도 Silver에는 영향 없음
    )