Lakehouse Architecture

from pyspark.sql import SparkSession
from delta import *

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

spark = configure_spark_with_delta_pip(builder).getOrCreate()

# Create sample data
data = spark.createDataFrame([
    (1, "Alice", 100),
    (2, "Bob", 200),
    (3, "Charlie", 300)
], ["id", "name", "amount"])

# Write as Delta table (ACID transaction)
data.write.format("delta").mode("overwrite").save("s3://my-bucket/delta/sales")

# Result: Atomic write, no partial data if failure occurs

# ============ CONCURRENT WRITES (ACID Isolation) ============
# Writer 1: Update Alice's amount
df1 = spark.read.format("delta").load("s3://my-bucket/delta/sales")
df1_updated = df1.filter("name != 'Alice'").union(
    spark.createDataFrame([(1, "Alice", 150)], ["id", "name", "amount"])
)
df1_updated.write.format("delta").mode("overwrite").save("s3://my-bucket/delta/sales")

# Writer 2: Simultaneously update Bob's amount (different transaction)
df2 = spark.read.format("delta").load("s3://my-bucket/delta/sales")
df2_updated = df2.filter("name != 'Bob'").union(
    spark.createDataFrame([(2, "Bob", 250)], ["id", "name", "amount"])
)
df2_updated.write.format("delta").mode("overwrite").save("s3://my-bucket/delta/sales")

# Result: ONE writer succeeds, the other retries with conflict resolution
# No corrupt data, no lost updates, no partial writes

# ============ MERGE (UPSERT) ============
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "s3://my-bucket/delta/sales")

# New data (updates + inserts)
updates = spark.createDataFrame([
    (1, "Alice", 175),     # Update existing
    (4, "Diana", 400)      # Insert new
], ["id", "name", "amount"])

# Atomic merge operation
delta_table.alias("target").merge(
    updates.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set={
    "amount": "source.amount"
}).whenNotMatchedInsert(values={
    "id": "source.id",
    "name": "source.name",
    "amount": "source.amount"
}).execute()

# Read result
spark.read.format("delta").load("s3://my-bucket/delta/sales").show()

# Output:
"""
+---+-------+------+
| id|   name|amount|
+---+-------+------+
|  1|  Alice|   175|  ← Updated
|  2|    Bob|   250|
|  3|Charlie|   300|
|  4|  Diana|   400|  ← Inserted
+---+-------+------+
"""

# ============ DELETE ============
delta_table.delete("id = 3")  # Atomic delete of Charlie

# Result: Charlie removed in single ACID transaction

# View table history
delta_table = DeltaTable.forPath(spark, "s3://my-bucket/delta/sales")
history = delta_table.history()
history.show()

# Output:
"""
+-------+-------------------+------+---------+---------+
|version|          timestamp|userId|operation|    stats|
+-------+-------------------+------+---------+---------+
|      3|2024-01-15 10:35:00|  user|  DELETE| {numR... |
|      2|2024-01-15 10:30:00|  user|   MERGE| {numU... |
|      1|2024-01-15 10:00:00|  user|   WRITE| {numF... |
|      0|2024-01-15 09:00:00|  user|   WRITE| {numF... |
+-------+-------------------+------+--------+----------+
"""

# ============ QUERY PREVIOUS VERSION (by version number) ============
# Read version 1 (before merge and delete)
df_v1 = spark.read.format("delta").option("versionAsOf", 1).load("s3://my-bucket/delta/sales")
df_v1.show()

"""
+---+-------+------+
| id|   name|amount|
+---+-------+------+
|  1|  Alice|   150|
|  2|    Bob|   250|
|  3|Charlie|   300|  ← Still exists in version 1
+---+-------+------+
"""

# ============ QUERY BY TIMESTAMP ============
# Read data as it was at 10:00 AM
df_at_time = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-15 10:00:00") \
    .load("s3://my-bucket/delta/sales")

# ============ ROLLBACK (RESTORE) ============
# Oops! Deleted wrong data. Rollback to version 2.
delta_table.restoreToVersion(2)

# Result: Table restored to state at version 2 (Charlie is back!)

# ============ USE CASES ============
# 1. Audit & Compliance: "Show me data as of Jan 1st for audit"
# 2. Rollback Mistakes: "Undo bad DELETE/UPDATE"
# 3. Reproducible ML: "Train model on data snapshot from last week"
# 4. A/B Testing: "Compare results before/after pipeline change"

# ============ VACUUM (Clean Old Versions) ============
# By default, Delta keeps all versions indefinitely
# Use VACUUM to delete old files (after retention period)
delta_table.vacuum(168)  # Delete files older than 168 hours (7 days)

# WARNING: After VACUUM, you can't time travel beyond retention period

# Original schema: id, name, amount
df_original = spark.read.format("delta").load("s3://my-bucket/delta/sales")
df_original.printSchema()
"""
root
 |-- id: integer
 |-- name: string
 |-- amount: integer
"""

# ============ ADD COLUMN ============
# New data with additional column
new_data = spark.createDataFrame([
    (5, "Eve", 500, "US"),     # Added 'country' column
    (6, "Frank", 600, "UK")
], ["id", "name", "amount", "country"])

# Write with schema merge enabled
new_data.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("s3://my-bucket/delta/sales")

# Read updated table
df_updated = spark.read.format("delta").load("s3://my-bucket/delta/sales")
df_updated.printSchema()
"""
root
 |-- id: integer
 |-- name: string
 |-- amount: integer
 |-- country: string  ← New column added!
"""

df_updated.show()
"""
+---+-------+------+-------+
| id|   name|amount|country|
+---+-------+------+-------+
|  1|  Alice|   175|   null|  ← Old rows have null for new column
|  2|    Bob|   250|   null|
|  4|  Diana|   400|   null|
|  5|    Eve|   500|     US|  ← New rows have values
|  6|  Frank|   600|     UK|
+---+-------+------+-------+
"""

# ============ RENAME COLUMN ============
# Delta Lake supports column mapping for safe renames
spark.sql("""
    ALTER TABLE delta.`s3://my-bucket/delta/sales`
    RENAME COLUMN amount TO total_amount
""")

# Existing queries using 'amount' will still work (backward compatible)

# ============ DROP COLUMN ============
spark.sql("""
    ALTER TABLE delta.`s3://my-bucket/delta/sales`
    DROP COLUMN country
""")

# ============ CHANGE DATA TYPE (with rewrite) ============
# Some changes require rewriting data
spark.sql("""
    ALTER TABLE delta.`s3://my-bucket/delta/sales`
    ALTER COLUMN total_amount TYPE DOUBLE
""")

# Result: Schema evolved without breaking existing readers!

from pyspark.sql import SparkSession

# Configure Spark with Iceberg
spark = SparkSession.builder \
    .appName("Iceberg") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.my_catalog.type", "hadoop") \
    .config("spark.sql.catalog.my_catalog.warehouse", "s3://my-bucket/iceberg") \
    .getOrCreate()

# Create Iceberg table
spark.sql("""
    CREATE TABLE my_catalog.db.sales (
        id INT,
        name STRING,
        amount DOUBLE,
        sale_date DATE
    )
    USING iceberg
    PARTITIONED BY (days(sale_date))
""")

# Insert data
spark.sql("""
    INSERT INTO my_catalog.db.sales VALUES
    (1, 'Alice', 100.0, DATE '2024-01-15'),
    (2, 'Bob', 200.0, DATE '2024-01-15'),
    (3, 'Charlie', 300.0, DATE '2024-01-16')
""")

# Read Iceberg table
df = spark.table("my_catalog.db.sales")
df.show()

# ============ SNAPSHOT MANAGEMENT ============
# View snapshots (similar to Delta versions)
spark.sql("SELECT * FROM my_catalog.db.sales.snapshots").show()

"""
+-------------------+------------+-------------------+
|   committed_at    | snapshot_id| parent_snapshot_id|
+-------------------+------------+-------------------+
|2024-01-15 10:00:00|    123456  |              null |
|2024-01-15 11:00:00|    123457  |            123456 |
+-------------------+------------+-------------------+
"""

# Time travel to specific snapshot
df_old = spark.read \
    .option("snapshot-id", "123456") \
    .table("my_catalog.db.sales")

# Time travel by timestamp
df_at_time = spark.read \
    .option("as-of-timestamp", "2024-01-15 10:30:00") \
    .table("my_catalog.db.sales")

# Original table: Partitioned by day
spark.sql("""
    CREATE TABLE my_catalog.db.events (
        event_id BIGINT,
        user_id INT,
        event_type STRING,
        timestamp TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(timestamp))
""")

# Insert 1 year of data (365 partitions)
# ... millions of rows ...

# PROBLEM: Queries filtering by user_id are slow (scan all partitions)

# ============ PARTITION EVOLUTION ============
# Add user_id to partitioning WITHOUT rewriting data!
spark.sql("""
    ALTER TABLE my_catalog.db.events
    ADD PARTITION FIELD bucket(100, user_id)
""")

# Now table is partitioned by BOTH:
# - days(timestamp) for old data
# - days(timestamp) + bucket(100, user_id) for new data

# NEW data written with updated partitioning
spark.sql("""
    INSERT INTO my_catalog.db.events VALUES
    (12345, 456, 'click', TIMESTAMP '2024-01-20 10:00:00')
""")

# Query benefits from BOTH partitioning schemes:
# - Old data: Filter by date
# - New data: Filter by date + user_id bucket

spark.sql("""
    SELECT * FROM my_catalog.db.events
    WHERE user_id = 456
      AND timestamp >= DATE '2024-01-01'
""").explain()

"""
Physical Plan:
  Scan iceberg partitions:
    - Old data: 365 partitions (by date only)
    - New data: 4 partitions (by date + user bucket)
  → 99% fewer files scanned for new data!
"""

# ============ REMOVE PARTITION FIELD ============
# Later, decide day-level partitioning is too granular
spark.sql("""
    ALTER TABLE my_catalog.db.events
    DROP PARTITION FIELD days(timestamp)
""")

# Add month-level partitioning instead
spark.sql("""
    ALTER TABLE my_catalog.db.events
    ADD PARTITION FIELD months(timestamp)
""")

# Result: Partitioning evolved over time without expensive rewrites!

# Traditional Hive partitioning (explicit partition columns)
spark.sql("""
    CREATE TABLE hive_sales (
        id INT,
        name STRING,
        amount DOUBLE
    )
    PARTITIONED BY (year INT, month INT, day INT)
""")

# Users MUST include partition columns in queries for performance
spark.sql("""
    SELECT * FROM hive_sales
    WHERE year = 2024 AND month = 1 AND day = 15  ← Must specify!
      AND amount > 100
""")

# PROBLEM: If user forgets partition filter, full table scan!


# Iceberg: Hidden partitioning (automatic pruning)
spark.sql("""
    CREATE TABLE my_catalog.db.iceberg_sales (
        id INT,
        name STRING,
        amount DOUBLE,
        sale_date DATE
    )
    USING iceberg
    PARTITIONED BY (days(sale_date))
""")

# Users query ONLY business columns (no partition awareness needed)
spark.sql("""
    SELECT * FROM my_catalog.db.iceberg_sales
    WHERE sale_date = DATE '2024-01-15'  ← Automatic partition pruning!
      AND amount > 100
""")

# Iceberg automatically:
# 1. Detects date filter
# 2. Prunes to relevant partition
# 3. Scans only necessary files

# Benefits:
# ✓ Simpler queries (no partition column maintenance)
# ✓ Impossible to forget partition filters
# ✓ Can change partitioning without breaking queries

from pyspark.sql import SparkSession

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

# ============ COPY-ON-WRITE TABLE ============
hudi_options_cow = {
    'hoodie.table.name': 'sales_cow',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.datasource.write.partitionpath.field': 'date',
    'hoodie.datasource.write.precombine.field': 'timestamp',
    'hoodie.datasource.write.table.type': 'COPY_ON_WRITE'  # CoW mode
}

df = spark.createDataFrame([
    (1, "Alice", 100, "2024-01-15", "2024-01-15T10:00:00"),
    (2, "Bob", 200, "2024-01-15", "2024-01-15T10:05:00")
], ["id", "name", "amount", "date", "timestamp"])

df.write.format("hudi") \
    .options(**hudi_options_cow) \
    .mode("overwrite") \
    .save("s3://my-bucket/hudi/sales_cow")

# ============ MERGE-ON-READ TABLE ============
hudi_options_mor = {
    'hoodie.table.name': 'sales_mor',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.datasource.write.partitionpath.field': 'date',
    'hoodie.datasource.write.precombine.field': 'timestamp',
    'hoodie.datasource.write.table.type': 'MERGE_ON_READ'  # MoR mode
}

df.write.format("hudi") \
    .options(**hudi_options_mor) \
    .mode("overwrite") \
    .save("s3://my-bucket/hudi/sales_mor")

# Initial load at 10:00 AM
initial_df = spark.createDataFrame([
    (1, "Alice", 100, "2024-01-15", "2024-01-15T10:00:00"),
    (2, "Bob", 200, "2024-01-15", "2024-01-15T10:00:00")
], ["id", "name", "amount", "date", "timestamp"])

initial_df.write.format("hudi").options(**hudi_options).mode("overwrite") \
    .save("s3://my-bucket/hudi/sales")

# Get commit time (for incremental queries)
commit_time_1 = spark.read.format("hudi") \
    .load("s3://my-bucket/hudi/sales") \
    .select("_hoodie_commit_time").first()[0]

print(f"Commit 1: {commit_time_1}")  # e.g., "20240115100000"

# ============ UPDATE at 11:00 AM ============
updates = spark.createDataFrame([
    (1, "Alice", 150, "2024-01-15", "2024-01-15T11:00:00"),  # Updated
    (3, "Charlie", 300, "2024-01-15", "2024-01-15T11:00:00") # New
], ["id", "name", "amount", "date", "timestamp"])

updates.write.format("hudi").options(**hudi_options).mode("append") \
    .save("s3://my-bucket/hudi/sales")

# ============ INCREMENTAL QUERY ============
# Read ONLY changes since commit_time_1
incremental_df = spark.read.format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", commit_time_1) \
    .load("s3://my-bucket/hudi/sales")

incremental_df.show()

"""
+---+-------+------+----------+-------------------+
| id|   name|amount|      date|          timestamp|
+---+-------+------+----------+-------------------+
|  1|  Alice|   150|2024-01-15|2024-01-15T11:00:00|  ← Updated
|  3|Charlie|   300|2024-01-15|2024-01-15T11:00:00|  ← New
+---+-------+------+----------+-------------------+

Bob not returned (unchanged since commit_time_1)
"""

# USE CASE: Incremental ETL pipeline
# Process only new/changed data, not entire table
def incremental_pipeline(last_commit_time):
    """Process only data changed since last run"""
    new_data = spark.read.format("hudi") \
        .option("hoodie.datasource.query.type", "incremental") \
        .option("hoodie.datasource.read.begin.instanttime", last_commit_time) \
        .load("s3://my-bucket/hudi/sales")

    # Transform only new data
    from pyspark.sql.functions import col
    transformed = new_data.withColumn("amount_doubled", col("amount") * 2)

    # Write to downstream system
    transformed.write.mode("append").save("s3://my-bucket/processed/")

    # Return latest commit time for next run
    return new_data.select("_hoodie_commit_time").first()[0]

# Run incrementally (only processes changes)
last_commit = "20240115100000"
last_commit = incremental_pipeline(last_commit)  # Process changes
last_commit = incremental_pipeline(last_commit)  # Process next batch (if any)