스키마 진화 심화

| 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에는 영향 없음
    )