# Traditional data lake
# What if Reader reads while Writer is writing?
df.write.parquet("s3://bucket/data/")  # Writing...
# Simultaneously in another process
df = spark.read.parquet("s3://bucket/data/")  # May read incomplete data

# Compatibility issues with existing files when adding columns
df_v1.write.parquet("data/v1/")  # 3 columns
df_v2.write.parquet("data/v2/")  # 4 columns
# Schema conflicts possible when reading both versions simultaneously

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. Create Delta table
df.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("date") \
    .save("s3://bucket/delta/events")

# 2. ACID transactions
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
# Read by specific version
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("s3://bucket/delta/events")

# Read by specific time
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 (multi-dimensional clustering)
delta_table.optimize().executeZOrderBy("user_id", "product_id")

# 3. Vacuum (delete old files)
delta_table.vacuum(168)  # Delete files older than 7 days

# 4. Data Skipping (statistics-based skip)
# Automatically collects and uses min/max statistics

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. Create Iceberg table
spark.sql("""
    CREATE TABLE events (
        id INT,
        name STRING,
        amount DOUBLE,
        event_time TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_time))
""")

# 2. Write data
df.writeTo("events").append()

# 3. ACID transactions
# 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
# By specific snapshot
df_snapshot = spark.read \
    .option("snapshot-id", 1234567890) \
    .format("iceberg") \
    .load("events")

# By specific time
df_timestamp = spark.read \
    .option("as-of-timestamp", "1705305600000") \
    .format("iceberg") \
    .load("events")

# 5. Schema evolution
spark.sql("ALTER TABLE events ADD COLUMN category STRING")

# 6. Partition Evolution (without rewriting existing data)
spark.sql("""
    ALTER TABLE events 
    REPLACE PARTITION FIELD days(event_time) 
    WITH hours(event_time)
""")

# 1. Expire Snapshots (clean old snapshots)
spark.sql("CALL spark_catalog.system.expire_snapshots('events', TIMESTAMP '2024-01-01 00:00:00')")

# 2. Remove Orphan Files (delete orphan files)
spark.sql("CALL spark_catalog.system.remove_orphan_files('events')")

# 3. Rewrite Data Files (merge small files)
spark.sql("CALL spark_catalog.system.rewrite_data_files('events')")

# 4. Rewrite Manifests (optimize manifests)
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. Create Hudi table (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 (core feature)
updates_df.write \
    .format("hudi") \
    .options(**hudi_options) \
    .mode("append") \
    .save("s3://bucket/hudi/events")

# 3. Incremental Query (incremental read)
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
# Data at specific time
df_snapshot = spark.read \
    .format("hudi") \
    .option("as.of.instant", "20240115120000") \
    .load("s3://bucket/hudi/events")

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

# 1. Clustering (file reorganization)
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'

# Load 1TB data into each format
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

# Upsert 1 million records
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(...)

# Query data from 7 days ago
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

# Read only changed data since last processing
# Hudi's powerful feature

# Hudi 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")

print(f"Changed records: {incremental_df.count():,}")
# Result: Changed records: 145,234 (0.14% of total)

# Using Delta Lake in Databricks
# 1. Unity Catalog integration
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 engine optimization
spark.conf.set("spark.databricks.photon.enabled", "true")

# Supports Spark, Presto, Flink, Athena
# 1. Create in Spark
spark.sql("""
    CREATE TABLE iceberg_catalog.db.events (
        id INT,
        name STRING,
        amount DOUBLE
    )
    USING iceberg
    PARTITIONED BY (days(event_time))
""")

# 2. Query in Presto
# SELECT * FROM iceberg.db.events WHERE date = DATE '2024-01-15'

# 3. Streaming write in 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 integration
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 pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

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

# 1. Read CDC events from Kafka
cdc_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "mysql.events") \
    .load()

# 2. Parse CDC events
parsed_df = cdc_df.select(
    from_json(col("value").cast("string"), cdc_schema).alias("data")
).select("data.*")

# 3. Streaming write to 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. Process downstream with incremental read
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")

# Large-scale batch 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")

# Query optimization with 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'
    )
""")

# Convert existing Parquet table to Delta Lake
from delta.tables import DeltaTable

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

# 2. Convert to new location
df = spark.read.parquet("s3://bucket/parquet/events")
df.write.format("delta").save("s3://bucket/delta/events")

# 3. Validation
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!"

# Migrate Parquet to Iceberg
# 1. Create Iceberg metadata on existing Parquet location
spark.sql("""
    CREATE TABLE iceberg_catalog.db.events
    USING iceberg
    LOCATION 's3://bucket/parquet/events'
    AS SELECT * FROM parquet.`s3://bucket/parquet/events`
""")

# 2. Or 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")

# Incremental migration by partition
from datetime import datetime, timedelta

def migrate_partition(source_format, target_format, date):
    """Migrate specific partition to new format"""
    
    # Read source
    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}'")
    
    # Write target
    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}")

# Migrate entire period
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)