Video Script: Delta Lake Essentials¶

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

# Create sample data
data = [
    (1, "Product A", "Electronics", 299.99, "2024-01-15"),
    (2, "Product B", "Clothing", 49.99, "2024-01-15"),
    (3, "Product C", "Electronics", 599.99, "2024-01-15"),
    (4, "Product D", "Home", 149.99, "2024-01-15"),
]

columns = ["product_id", "product_name", "category", "price", "created_date"]

df = spark.createDataFrame(data, columns)

# Write as Delta table
delta_path = "abfss://delta@mydatalake.dfs.core.windows.net/products"

df.write \
    .format("delta") \
    .mode("overwrite") \
    .save(delta_path)

print("✅ Delta table created successfully!")

# Create new records
new_products = [
    (5, "Product E", "Electronics", 799.99, "2024-01-16"),
    (6, "Product F", "Sports", 89.99, "2024-01-16"),
]

new_df = spark.createDataFrame(new_products, columns)

# Append to Delta table
new_df.write \
    .format("delta") \
    .mode("append") \
    .save(delta_path)

# Read and verify
products_df = spark.read.format("delta").load(delta_path)
print(f"Total products: {products_df.count()}")
display(products_df.orderBy("product_id"))

# Add new column to data
evolved_data = [
    (7, "Product G", "Books", 24.99, "2024-01-17", "In Stock"),  # New column
    (8, "Product H", "Books", 19.99, "2024-01-17", "Low Stock"),
]

evolved_columns = columns + ["stock_status"]
evolved_df = spark.createDataFrame(evolved_data, evolved_columns)

# Write with schema merge
evolved_df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save(delta_path)

# Verify schema evolution
products_df = spark.read.format("delta").load(delta_path)
products_df.printSchema()
display(products_df)

# Create managed table (metadata + data in metastore)
df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .partitionBy("category") \
    .saveAsTable("products_catalog")

# Query using SQL
spark.sql("""
    SELECT
        category,
        COUNT(*) as product_count,
        AVG(price) as avg_price,
        MIN(price) as min_price,
        MAX(price) as max_price
    FROM products_catalog
    GROUP BY category
    ORDER BY avg_price DESC
""").show()

# Table is now accessible from Serverless SQL too!

from delta.tables import DeltaTable

# Load Delta table
delta_table = DeltaTable.forPath(spark, delta_path)

# Update prices for Electronics category
delta_table.update(
    condition = "category = 'Electronics'",
    set = {
        "price": "price * 0.9",  # 10% discount
        "updated_date": "current_date()"
    }
)

# Verify updates
spark.read.format("delta").load(delta_path) \
    .filter("category = 'Electronics'") \
    .select("product_name", "price", "category") \
    .show()

# Delete products by condition
delta_table.delete("price < 50")

# Verify deletion
current_products = spark.read.format("delta").load(delta_path)
print(f"Products after deletion: {current_products.count()}")

# Note: Deleted data is still in Delta log for time travel

# Create updates and new records
updates_data = [
    (1, "Product A v2", "Electronics", 279.99, "2024-01-18", "In Stock"),  # Update
    (3, "Product C v2", "Electronics", 549.99, "2024-01-18", "In Stock"),  # Update
    (9, "Product I", "Gaming", 399.99, "2024-01-18", "Pre-Order"),         # Insert
    (10, "Product J", "Gaming", 449.99, "2024-01-18", "Pre-Order"),        # Insert
]

updates_df = spark.createDataFrame(updates_data, evolved_columns)

# Perform MERGE
delta_table.alias("target").merge(
    updates_df.alias("source"),
    "target.product_id = source.product_id"
) \
.whenMatchedUpdate(set = {
    "product_name": "source.product_name",
    "category": "source.category",
    "price": "source.price",
    "created_date": "source.created_date",
    "stock_status": "source.stock_status"
}) \
.whenNotMatchedInsert(values = {
    "product_id": "source.product_id",
    "product_name": "source.product_name",
    "category": "source.category",
    "price": "source.price",
    "created_date": "source.created_date",
    "stock_status": "source.stock_status"
}) \
.execute()

print("✅ MERGE operation completed!")

# View results
display(spark.read.format("delta").load(delta_path).orderBy("product_id"))

# Simulate CDC records
cdc_records = [
    (1, "Product A v3", "Electronics", 259.99, "2024-01-19", "In Stock", "UPDATE"),
    (2, "Product B", "Clothing", 0, "2024-01-19", None, "DELETE"),
    (11, "Product K", "Toys", 34.99, "2024-01-19", "In Stock", "INSERT"),
]

cdc_columns = evolved_columns + ["operation"]
cdc_df = spark.createDataFrame(cdc_records, cdc_columns)

# Apply CDC operations
delta_table.alias("target").merge(
    cdc_df.alias("cdc"),
    "target.product_id = cdc.product_id"
) \
.whenMatchedUpdate(
    condition = "cdc.operation = 'UPDATE'",
    set = {
        "product_name": "cdc.product_name",
        "price": "cdc.price",
        "stock_status": "cdc.stock_status"
    }
) \
.whenMatchedDelete(
    condition = "cdc.operation = 'DELETE'"
) \
.whenNotMatchedInsert(
    condition = "cdc.operation = 'INSERT'",
    values = {
        "product_id": "cdc.product_id",
        "product_name": "cdc.product_name",
        "category": "cdc.category",
        "price": "cdc.price",
        "stock_status": "cdc.stock_status"
    }
) \
.execute()

print("✅ CDC applied successfully!")

# View Delta table history
history_df = delta_table.history()

# Show recent operations
display(
    history_df.select(
        "version",
        "timestamp",
        "operation",
        "operationParameters",
        "operationMetrics"
    ).orderBy(desc("version"))
)

# Get specific version details
latest_version = history_df.select(max("version")).collect()[0][0]
print(f"Latest version: {latest_version}")

# Query specific version
version_0 = spark.read.format("delta").option("versionAsOf", 0).load(delta_path)
print(f"Version 0 count: {version_0.count()}")
display(version_0)

# Query as of specific timestamp
from datetime import datetime, timedelta
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")

df_yesterday = spark.read.format("delta") \
    .option("timestampAsOf", yesterday) \
    .load(delta_path)

# Compare versions
print(f"Current version: {spark.read.format('delta').load(delta_path).count()} records")
print(f"Yesterday: {df_yesterday.count()} records")

# Find differences
current = spark.read.format("delta").load(delta_path)
diff = current.subtract(df_yesterday)
print(f"Records added since yesterday: {diff.count()}")
display(diff)

# Oops! Let's rollback to version 5
delta_table.restoreToVersion(5)

print("✅ Restored to version 5")

# Or restore to specific timestamp
# delta_table.restoreToTimestamp("2024-01-15 10:00:00")

# Verify restoration
current = spark.read.format("delta").load(delta_path)
print(f"Current record count: {current.count()}")

# View history shows RESTORE operation
display(delta_table.history().limit(5))

# Check current file structure
file_stats = spark.sql(f"""
    SELECT
        COUNT(*) as file_count,
        SUM(size) / 1024 / 1024 as total_mb,
        AVG(size) / 1024 / 1024 as avg_file_mb
    FROM delta.`{delta_path}`.files
""")

print("Before OPTIMIZE:")
display(file_stats)

# Run OPTIMIZE
from delta.tables import DeltaTable
delta_table.optimize().executeCompaction()

# Check after optimization
file_stats_after = spark.sql(f"""
    SELECT
        COUNT(*) as file_count,
        SUM(size) / 1024 / 1024 as total_mb,
        AVG(size) / 1024 / 1024 as avg_file_mb
    FROM delta.`{delta_path}`.files
""")

print("After OPTIMIZE:")
display(file_stats_after)

# Z-ORDER by frequently queried columns
delta_table.optimize().executeZOrderBy("category", "price")

print("✅ Z-ORDER optimization completed!")

# Test query performance
import time

# Query before Z-ORDER (use earlier version)
start = time.time()
spark.read.format("delta").option("versionAsOf", 0).load(delta_path) \
    .filter("category = 'Electronics' AND price > 500") \
    .count()
before_time = time.time() - start

# Query after Z-ORDER
start = time.time()
spark.read.format("delta").load(delta_path) \
    .filter("category = 'Electronics' AND price > 500") \
    .count()
after_time = time.time() - start

print(f"Before Z-ORDER: {before_time:.2f} seconds")
print(f"After Z-ORDER: {after_time:.2f} seconds")
print(f"Performance improvement: {((before_time - after_time) / before_time * 100):.1f}%")

# Check retention period (default: 7 days)
spark.conf.get("spark.databricks.delta.retentionDurationCheck.enabled")

# Set retention to 0 for demo (DON'T DO THIS IN PRODUCTION!)
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

# VACUUM removes files older than retention period
delta_table.vacuum(0)  # Remove immediately (for demo only)

print("✅ VACUUM completed!")

# Note: Time travel won't work for vacuumed versions
# In production, keep default 7-day retention or longer

# 1. Table Design
✅ Partition large tables by date or high-cardinality column
✅ Use appropriate data types (avoid STRING for everything)
✅ Define schema explicitly rather than inferring
✅ Enable auto-optimize for write-heavy tables

# 2. Performance
✅ Run OPTIMIZE weekly on active tables
✅ Use Z-ORDER on frequently filtered columns
✅ Keep partition sizes between 100MB-1GB
✅ Cache frequently accessed data

# 3. Data Quality
✅ Enforce schema with mergeSchema=false (default)
✅ Use CHECK constraints for data validation
✅ Implement proper error handling in MERGE operations
✅ Test CDC patterns thoroughly

# 4. Operations
✅ Keep VACUUM retention at 7+ days minimum
✅ Monitor table history and file counts
✅ Use DESCRIBE HISTORY to audit changes
✅ Implement proper access controls

# 5. Cost Optimization
✅ Compress data with Snappy or GZIP
✅ Archive cold data to separate tables
✅ Monitor storage costs in Azure Portal
✅ Clean up unused tables regularly