# 전통적인 데이터 레이크
# Writer가 파일 쓰는 중에 Reader가 읽으면?
df.write.parquet("s3://bucket/data/")  # 쓰기 중...
# 동시에 다른 프로세스
df = spark.read.parquet("s3://bucket/data/")  # 불완전한 데이터 읽을 수 있음

# 컬럼 추가 시 기존 파일과 호환성 문제
df_v1.write.parquet("data/v1/")  # 컬럼 3개
df_v2.write.parquet("data/v2/")  # 컬럼 4개
# 두 버전 동시 읽기 시 스키마 충돌 가능

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Delta Lake") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 1. Delta 테이블 생성
df.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("date") \
    .save("s3://bucket/delta/events")

# 2. ACID 트랜잭션
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "s3://bucket/delta/events")

# Update
delta_table.update(
    condition = "date = '2024-01-15'",
    set = {"amount": "amount * 1.1"}
)

# Delete
delta_table.delete("amount < 0")

# Merge (Upsert)
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    set = {"amount": "source.amount"}
).whenNotMatchedInsert(
    values = {"id": "source.id", "amount": "source.amount"}
).execute()

# 3. Time Travel
# 특정 버전으로 읽기
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("s3://bucket/delta/events")

# 특정 시간으로 읽기
df_yesterday = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-14 00:00:00") \
    .load("s3://bucket/delta/events")

# 4. Schema Evolution
df_new_schema.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("s3://bucket/delta/events")

# 1. Optimize (Compaction)
delta_table.optimize().executeCompaction()

# 2. Z-Ordering (다차원 클러스터링)
delta_table.optimize().executeZOrderBy("user_id", "product_id")

# 3. Vacuum (오래된 파일 삭제)
delta_table.vacuum(168)  # 7일 이상 된 파일 삭제

# 4. Data Skipping (통계 기반 스킵)
# 자동으로 min/max 통계 수집 및 활용

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Iceberg") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.spark_catalog.type", "hadoop") \
    .config("spark.sql.catalog.spark_catalog.warehouse", "s3://bucket/warehouse") \
    .getOrCreate()

# 1. Iceberg 테이블 생성
spark.sql("""
    CREATE TABLE events (
        id INT,
        name STRING,
        amount DOUBLE,
        event_time TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_time))
""")

# 2. 데이터 쓰기
df.writeTo("events").append()

# 3. ACID 트랜잭션
# Merge (Upsert)
spark.sql("""
    MERGE INTO events t
    USING updates s
    ON t.id = s.id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

# Delete
spark.sql("DELETE FROM events WHERE amount < 0")

# 4. Time Travel
# 특정 스냅샷
df_snapshot = spark.read \
    .option("snapshot-id", 1234567890) \
    .format("iceberg") \
    .load("events")

# 특정 시간
df_timestamp = spark.read \
    .option("as-of-timestamp", "1705305600000") \
    .format("iceberg") \
    .load("events")

# 5. 스키마 진화
spark.sql("ALTER TABLE events ADD COLUMN category STRING")

# 6. Partition Evolution (기존 데이터 재작성 없이)
spark.sql("""
    ALTER TABLE events 
    REPLACE PARTITION FIELD days(event_time) 
    WITH hours(event_time)
""")

# 1. Expire Snapshots (오래된 스냅샷 정리)
spark.sql("CALL spark_catalog.system.expire_snapshots('events', TIMESTAMP '2024-01-01 00:00:00')")

# 2. Remove Orphan Files (고아 파일 삭제)
spark.sql("CALL spark_catalog.system.remove_orphan_files('events')")

# 3. Rewrite Data Files (작은 파일 병합)
spark.sql("CALL spark_catalog.system.rewrite_data_files('events')")

# 4. Rewrite Manifests (manifest 최적화)
spark.sql("CALL spark_catalog.system.rewrite_manifests('events')")

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Hudi") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .getOrCreate()

# 1. Hudi 테이블 생성 (Copy on Write)
hudi_options = {
    'hoodie.table.name': 'events',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.datasource.write.precombine.field': 'event_time',
    'hoodie.datasource.write.partitionpath.field': 'date',
    'hoodie.datasource.write.table.type': 'COPY_ON_WRITE',
    'hoodie.datasource.write.operation': 'upsert'
}

df.write \
    .format("hudi") \
    .options(**hudi_options) \
    .mode("overwrite") \
    .save("s3://bucket/hudi/events")

# 2. Upsert (핵심 기능)
updates_df.write \
    .format("hudi") \
    .options(**hudi_options) \
    .mode("append") \
    .save("s3://bucket/hudi/events")

# 3. Incremental Query (증분 읽기)
incremental_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", "20240115120000") \
    .option("hoodie.datasource.read.end.instanttime", "20240115130000") \
    .load("s3://bucket/hudi/events")

# 4. Time Travel
# 특정 시간의 데이터
df_snapshot = spark.read \
    .format("hudi") \
    .option("as.of.instant", "20240115120000") \
    .load("s3://bucket/hudi/events")

# 5. Compaction (MoR에서 중요)
spark.sql("""
    CALL run_compaction(
        table => 'events',
        path => 's3://bucket/hudi/events'
    )
""")

# 1. Clustering (파일 재구성)
hudi_options['hoodie.clustering.inline'] = 'true'
hudi_options['hoodie.clustering.inline.max.commits'] = '4'

# 2. Indexing
hudi_options['hoodie.index.type'] = 'BLOOM'  # BLOOM, SIMPLE, GLOBAL_BLOOM

# 3. File Sizing
hudi_options['hoodie.parquet.small.file.limit'] = '104857600'  # 100MB
hudi_options['hoodie.parquet.max.file.size'] = '134217728'     # 128MB

# 4. Async Compaction
hudi_options['hoodie.compact.inline'] = 'false'
hudi_options['hoodie.compact.schedule.inline'] = 'true'

# 1TB 데이터를 각 포맷으로 적재
import time

# Delta Lake
start = time.time()
df.write.format("delta").mode("overwrite").save("s3://bucket/delta/")
delta_time = time.time() - start

# Iceberg
start = time.time()
df.writeTo("iceberg_table").create()
iceberg_time = time.time() - start

# Hudi (CoW)
start = time.time()
df.write.format("hudi").options(**hudi_cow_options).save("s3://bucket/hudi_cow/")
hudi_cow_time = time.time() - start

# Hudi (MoR)
start = time.time()
df.write.format("hudi").options(**hudi_mor_options).save("s3://bucket/hudi_mor/")
hudi_mor_time = time.time() - start

# 100만 레코드 upsert
updates_df = spark.read.parquet("s3://bucket/updates/")

# Delta Lake
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set = {...}).whenNotMatchedInsert(values = {...}).execute()

# Iceberg
spark.sql("MERGE INTO events ...")

# Hudi
updates_df.write.format("hudi").options(**hudi_options).mode("append").save(...)

# 7일 전 데이터 조회
import time

# Delta Lake
start = time.time()
df = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-08 00:00:00") \
    .load("s3://bucket/delta/events")
count = df.count()
delta_tt_time = time.time() - start

# Iceberg
start = time.time()
df = spark.read.format("iceberg") \
    .option("as-of-timestamp", "2024-01-08 00:00:00") \
    .load("events")
count = df.count()
iceberg_tt_time = time.time() - start

# Hudi
start = time.time()
df = spark.read.format("hudi") \
    .option("as.of.instant", "20240108000000") \
    .load("s3://bucket/hudi/events")
count = df.count()
hudi_tt_time = time.time() - start

# 마지막 처리 이후 변경된 데이터만 읽기
# Hudi의 강력한 기능

# Hudi Incremental Query
incremental_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", "20240115120000") \
    .load("s3://bucket/hudi/events")

print(f"Changed records: {incremental_df.count():,}")
# 결과: Changed records: 145,234 (전체의 0.14%)

# Databricks에서 Delta Lake 사용
# 1. Unity Catalog 통합
spark.sql("""
    CREATE TABLE main.analytics.events
    USING DELTA
    PARTITIONED BY (date)
    LOCATION 's3://bucket/delta/events'
""")

# 2. Delta Live Tables (DLT)
@dlt.table(
    name="events_gold",
    comment="Cleansed events table"
)
def events_gold():
    return (
        dlt.read("events_silver")
        .where("amount > 0")
        .select("id", "name", "amount", "date")
    )

# 3. Photon 엔진 최적화
spark.conf.set("spark.databricks.photon.enabled", "true")

# Spark, Presto, Flink, Athena 모두 지원
# 1. Spark에서 생성
spark.sql("""
    CREATE TABLE iceberg_catalog.db.events (
        id INT,
        name STRING,
        amount DOUBLE
    )
    USING iceberg
    PARTITIONED BY (days(event_time))
""")

# 2. Presto에서 쿼리
# SELECT * FROM iceberg.db.events WHERE date = DATE '2024-01-15'

# 3. Flink에서 스트리밍 쓰기
tableEnv.executeSql("""
    CREATE TABLE events (
        id INT,
        name STRING,
        amount DOUBLE,
        event_time TIMESTAMP(3)
    ) WITH (
        'connector' = 'iceberg',
        'catalog-name' = 'iceberg_catalog'
    )
""")

# 4. AWS Glue Catalog 통합
spark.conf.set("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.glue_catalog.warehouse", "s3://bucket/warehouse")
spark.conf.set("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")

# Kafka CDC → Hudi MoR 파이프라인
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("CDC to Hudi") \
    .getOrCreate()

# 1. Kafka에서 CDC 이벤트 읽기
cdc_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "mysql.events") \
    .load()

# 2. CDC 이벤트 파싱
parsed_df = cdc_df.select(
    from_json(col("value").cast("string"), cdc_schema).alias("data")
).select("data.*")

# 3. Hudi MoR로 스트리밍 쓰기
hudi_options = {
    'hoodie.table.name': 'events',
    'hoodie.datasource.write.table.type': 'MERGE_ON_READ',
    'hoodie.datasource.write.operation': 'upsert',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.datasource.write.precombine.field': 'updated_at',
    'hoodie.compact.inline': 'false',
    'hoodie.compact.schedule.inline': 'true'
}

parsed_df.writeStream \
    .format("hudi") \
    .options(**hudi_options) \
    .outputMode("append") \
    .option("checkpointLocation", "s3://bucket/checkpoints/") \
    .start("s3://bucket/hudi/events")

# 4. 증분 읽기로 downstream 처리
incremental_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", last_commit_time) \
    .load("s3://bucket/hudi/events")

# 대규모 배치 ETL
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Batch Analytics") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# Delta Lake
delta_table = DeltaTable.forPath(spark, "s3://bucket/delta/events")

# Z-Ordering으로 쿼리 최적화
delta_table.optimize() \
    .where("date >= '2024-01-01'") \
    .executeZOrderBy("user_id", "product_id")

# Iceberg
spark.sql("""
    CALL spark_catalog.system.rewrite_data_files(
        table => 'events',
        strategy => 'sort',
        sort_order => 'user_id, product_id'
    )
""")

# 기존 Parquet 테이블을 Delta Lake로 변환
from delta.tables import DeltaTable

# 1. In-place 변환
DeltaTable.convertToDelta(
    spark,
    "parquet.`s3://bucket/events`",
    "date STRING"
)

# 2. 새로운 위치로 변환
df = spark.read.parquet("s3://bucket/parquet/events")
df.write.format("delta").save("s3://bucket/delta/events")

# 3. 검증
delta_df = spark.read.format("delta").load("s3://bucket/delta/events")
parquet_df = spark.read.parquet("s3://bucket/parquet/events")

assert delta_df.count() == parquet_df.count(), "Count mismatch!"

# Parquet를 Iceberg로 마이그레이션
# 1. 기존 Parquet 위치에 Iceberg 메타데이터 생성
spark.sql("""
    CREATE TABLE iceberg_catalog.db.events
    USING iceberg
    LOCATION 's3://bucket/parquet/events'
    AS SELECT * FROM parquet.`s3://bucket/parquet/events`
""")

# 2. 또는 CTAS (Create Table As Select)
spark.sql("""
    CREATE TABLE iceberg_catalog.db.events
    USING iceberg
    PARTITIONED BY (date)
    AS SELECT * FROM parquet.`s3://bucket/parquet/events`
""")

# Delta Lake → Iceberg
delta_df = spark.read.format("delta").load("s3://bucket/delta/events")
delta_df.writeTo("iceberg_catalog.db.events").create()

# Iceberg → Delta Lake
iceberg_df = spark.read.format("iceberg").load("iceberg_catalog.db.events")
iceberg_df.write.format("delta").save("s3://bucket/delta/events")

# 파티션별 점진적 마이그레이션
from datetime import datetime, timedelta

def migrate_partition(source_format, target_format, date):
    """특정 파티션을 새로운 포맷으로 마이그레이션"""
    
    # 소스 읽기
    if source_format == "parquet":
        df = spark.read.parquet(f"s3://bucket/parquet/events/date={date}/")
    elif source_format == "delta":
        df = spark.read.format("delta").load(f"s3://bucket/delta/events") \
            .where(f"date = '{date}'")
    
    # 타겟 쓰기
    if target_format == "iceberg":
        df.writeTo(f"iceberg_catalog.db.events").append()
    elif target_format == "hudi":
        df.write.format("hudi").options(**hudi_options).mode("append") \
            .save("s3://bucket/hudi/events")
    
    print(f"✓ Migrated: {date}")

# 전체 기간 마이그레이션
start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 12, 31)
current_date = start_date

while current_date <= end_date:
    date_str = current_date.strftime("%Y-%m-%d")
    try:
        migrate_partition("parquet", "iceberg", date_str)
    except Exception as e:
        print(f"✗ Failed: {date_str} - {e}")
    
    current_date += timedelta(days=1)