Delta Lake

# Delta Lake 핵심 기능 예제

from delta.tables import DeltaTable
from pyspark.sql.functions import col, current_timestamp

# 1. Delta 테이블 생성
df = spark.createDataFrame([
    ("user1", "Alice", 100),
    ("user2", "Bob", 200),
    ("user3", "Charlie", 150)
], ["user_id", "name", "balance"])

df.write.format("delta").mode("overwrite").save("/delta/accounts")

# 2. ACID 트랜잭션 - MERGE (Upsert)
updates = spark.createDataFrame([
    ("user1", "Alice", 120),   # 기존 사용자 업데이트
    ("user4", "David", 300)    # 새 사용자 삽입
], ["user_id", "name", "balance"])

delta_table = DeltaTable.forPath(spark, "/delta/accounts")

delta_table.alias("target").merge(
    updates.alias("source"),
    "target.user_id = source.user_id"
).whenMatchedUpdate(set={
    "name": col("source.name"),
    "balance": col("source.balance")
}).whenNotMatchedInsert(values={
    "user_id": col("source.user_id"),
    "name": col("source.name"),
    "balance": col("source.balance")
}).execute()

# 3. Time Travel - 버전으로 조회
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/delta/accounts")
print("버전 0 (초기 상태):")
df_v0.show()

# 타임스탬프로 조회
df_past = spark.read.format("delta").option("timestampAsOf", "2024-01-15 10:00:00").load("/delta/accounts")

# 4. 히스토리 조회
delta_table.history().show(truncate=False)

# 5. 롤백 - 이전 버전으로 복원
# 방법 1: RESTORE 명령
spark.sql("RESTORE TABLE delta.`/delta/accounts` TO VERSION AS OF 0")

# 방법 2: 이전 버전 덮어쓰기
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/delta/accounts")
df_v0.write.format("delta").mode("overwrite").save("/delta/accounts")

# 6. 스키마 진화 - 새 컬럼 자동 추가
new_data = spark.createDataFrame([
    ("user5", "Eve", 250, "VIP")
], ["user_id", "name", "balance", "tier"])

new_data.write.format("delta").mode("append").option("mergeSchema", "true").save("/delta/accounts")

# 7. 최적화 - 소규모 파일 병합
delta_table.optimize().executeCompaction()

# Z-ORDER로 쿼리 성능 최적화
delta_table.optimize().executeZOrderBy("user_id")

# 8. VACUUM - 오래된 파일 정리 (기본 7일 이상 지난 파일)
delta_table.vacuum(168)  # 168시간 = 7일

print("최종 테이블 상태:")
spark.read.format("delta").load("/delta/accounts").show()