Optimize Storage and Performance with Delta Lake

Level: Intermediate / Advanced | Platform: Azure Databricks Premium

Table of Contents

1. Data Lakehouse Architecture and Delta Lake
2. Storage Formats Compared
3. Delta Tables and Transaction Log
4. ACID Properties in Delta Lake
5. Creating and Managing Delta Tables
6. Loading Data from ADLS Gen2
7. Streaming to Delta Tables
8. Schema Enforcement and Evolution
9. Delta Lake Optimization Techniques
10. Partitioning — Directory-based Organization
11. Z-Order — Data Co-location
12. OPTIMIZE — Small File Compaction
13. Liquid Clustering — Adaptive Partitioning
14. Delta Caching — Local Acceleration
15. Data Skipping and Column Statistics
16. Photon Acceleration
17. Time Travel and Versioning
18. VACUUM — Cleaning Up Obsolete Files
19. Monitoring Delta Tables
20. Optimistic Concurrency Control
21. Advanced Azure Use Cases
22. Summary and Best Practices
23. Glossary

1. Data Lakehouse Architecture and Delta Lake

1.1 Evolution of Data Architectures

timeline
    title Evolution of Data Storage Architectures
    
    "Before 2010" : Data Warehouses only
                  : Structured data
                  : Expensive, but fast
                  : Teradata, Oracle DW
    
    "2010-2018" : Two-Tier Architecture
               : Data Lake (raw, cheap) + Data Warehouse (structured, fast)
               : HDFS, S3, ADLS
               : Complex ETL pipelines between the two
    
    "2018-present" : Lakehouse Architecture
                   : Delta Lake combines both worlds
                   : ACID on Data Lake
                   : ADLS Gen2 + Delta Lake + Databricks

1.2 Problems with the Two-Tier Architecture

graph LR
    subgraph "Two-Tier Architecture (legacy)"
        Sources[Data Sources] -->|Raw ingestion| Lake[Data Lake\nADLS Gen2\nLow cost, flexible]
        Lake -->|Complex ETL| DW[Data Warehouse\nAzure Synapse\nHigh perf, costly]
        
        DW --> BI[BI/Reporting]
        Lake --> ML[Data Science/ML]
        
        ETL[ETL Pipeline\nRedundancy!\nMaintainability!]
        Lake -.->|Duplication| DW
    end
    
    style ETL fill:#ff9999
    style Lake fill:#fff3e0
    style DW fill:#e8f5e9

Problems:

• Data duplication between the Lake and the Warehouse
• Complex ETL to maintain for synchronizing both
• Two separate systems with different APIs
• Double costs — pay for the Lake AND the Warehouse
• Latency — raw data is not immediately available

1.3 Lakehouse Architecture with Delta Lake

graph LR
    Sources[Data Sources] --> ADLS[ADLS Gen2\nUnified storage]
    ADLS --> Delta[Delta Lake\nMetadata layer]
    
    Delta --> APIs[Unified APIs]
    APIs --> SQL2[SQL Analytics\nPower BI, Synapse]
    APIs --> Spark[Spark DataFrame\nData Science, ML]
    APIs --> Stream[Streaming\nReal-time]
    
    style Delta fill:#e8f5e9,stroke:#4caf50
    note1[✅ Single system\n✅ ACID on the Lake\n✅ Warehouse performance\n✅ Lake cost]

2. Storage Formats Compared

2.1 CSV vs Parquet vs Delta Lake

2.2 Anatomy of a Delta Table on ADLS Gen2

abfss://container@account.dfs.core.windows.net/tables/customers/
│
├── _delta_log/                    ← Transaction Log
│   ├── 00000000000000000000.json  ← Version 0: CREATE TABLE
│   ├── 00000000000000000001.json  ← Version 1: INSERT
│   ├── 00000000000000000002.json  ← Version 2: UPDATE
│   └── 00000000000000000010.checkpoint.parquet ← Checkpoint every 10 ops
│
├── part-00001-xxxxx.snappy.parquet  ← Parquet data (partitions)
├── part-00002-xxxxx.snappy.parquet
└── country=USA/                    ← If partitioned by country
    └── part-00001-xxxx.snappy.parquet

3. Delta Tables and Transaction Log

3.1 How the DeltaLog Works

sequenceDiagram
    participant User as User
    participant Spark as Apache Spark
    participant Log as _delta_log/
    participant Storage as Parquet Files

    User->>Spark: INSERT 2 records
    Spark->>Storage: Write file-01.parquet
    Spark->>Log: Commit: {add: file-01.parquet, schema, metadata}
    Log-->>User: ✅ Version 0 confirmed

    User->>Spark: INSERT 2 more records
    Spark->>Storage: Write file-02.parquet
    Spark->>Log: Commit: {add: file-02.parquet}
    Log-->>User: ✅ Version 1 confirmed

    User->>Spark: DELETE record id=1
    Spark->>Storage: Write file-03.parquet (without id=1)
    Spark->>Log: Commit: {remove: file-01.parquet, add: file-03.parquet}
    Log-->>User: ✅ Version 2 confirmed
    Note over Storage: file-01.parquet still present (for Time Travel!)

3.2 JSON Commit Structure in the DeltaLog

{
  "commitInfo": {
    "timestamp": 1704067200000,
    "operation": "WRITE",
    "operationParameters": {
      "mode": "Append",
      "partitionBy": "[]"
    },
    "isBlindAppend": true,
    "operationMetrics": {
      "numFiles": "2",
      "numOutputRows": "150000",
      "numOutputBytes": "12582912"
    }
  },
  "metaData": {
    "schemaString": "{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\"},{\"name\":\"name\",\"type\":\"string\"}]}",
    "partitionColumns": []
  },
  "add": {
    "path": "part-00001-xxxx.snappy.parquet",
    "partitionValues": {},
    "size": 6291456,
    "stats": "{\"numRecords\":75000,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":75000}}"
  }
}

4. ACID Properties in Delta Lake

4.1 The Four ACID Properties

graph TB
    ACID[ACID Properties\nDelta Lake] --> A[Atomicity\nAll or nothing]
    ACID --> C[Consistency\nValid state guaranteed]
    ACID --> I[Isolation\nIndependent transactions]
    ACID --> D[Durability\nPersistent changes]
    
    A --> A1[If write fails\nno data is corrupted]
    C --> C1[Schema enforcement\nConstraints verified]
    I --> I1[Optimistic Concurrency\nNo blocking locks]
    D --> D1[Transaction log\nsurvives failures]

4.2 Optimistic Concurrency Control in Detail

Delta Lake uses Optimistic Concurrency Control (OCC), different from pessimistic locking:

# Demonstrating conflict handling with Delta Lake
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql import functions as F

spark = SparkSession.builder.getOrCreate()

# Create a table to demonstrate OCC
inventory_data = [(1, "Laptop", 999.99), (2, "Mouse", 29.99), (3, "Keyboard", 79.99)]
inventory_df = spark.createDataFrame(inventory_data, ["id", "name", "price"])

inventory_df.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("default.inventory_occ_demo")

# Transaction 1 and Transaction 2 run in parallel (simulated)
# Transaction 1: Increase all prices by 10%
delta_table = DeltaTable.forName(spark, "default.inventory_occ_demo")

# If two transactions try to modify the same rows simultaneously,
# Delta Lake detects the conflict and one of them fails with:
# "A newer version of the Delta table exists"
# The application must handle this by retrying the transaction

try:
    delta_table.update(
        condition=None,
        set={"price": F.col("price") * F.lit(1.1)}
    )
    print("✅ Price update successful")
except Exception as e:
    print(f"❌ Conflict detected: {e}")
    print("   Action: Re-read the table and retry the transaction")

5. Creating and Managing Delta Tables

5.1 Methods for Creating a Delta Table

-- Method 1: CREATE TABLE via SQL (Unity Catalog)
CREATE TABLE IF NOT EXISTS retailcatalog.sales.retail_customers (
    customer_id     INTEGER NOT NULL,
    first_name      STRING NOT NULL,
    last_name       STRING NOT NULL,
    account_balance DOUBLE,
    account_type    STRING,
    open_date       DATE,
    country         STRING
)
USING DELTA
PARTITIONED BY (country)
COMMENT 'Retail customer data';

-- Method 2: CTAS (Create Table As Select)
CREATE OR REPLACE TABLE retailcatalog.sales.premium_customers
AS
SELECT *
FROM retailcatalog.sales.retail_customers
WHERE account_balance > 50000;

-- Method 3: Via COPY INTO (loading from files)
COPY INTO retailcatalog.sales.retail_customers
FROM 'abfss://raw@retaildatalake.dfs.core.windows.net/customers/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

-- Verify the table
DESCRIBE TABLE EXTENDED retailcatalog.sales.retail_customers;
DESCRIBE HISTORY retailcatalog.sales.retail_customers;

# Method 4: Via PySpark DataFrame
from pyspark.sql import functions as F
from pyspark.sql.types import *

# Create from a DataFrame
customer_schema = StructType([
    StructField("customer_id",     IntegerType(), nullable=False),
    StructField("first_name",      StringType(),  nullable=False),
    StructField("last_name",       StringType(),  nullable=False),
    StructField("account_balance", DoubleType(),  nullable=True),
    StructField("account_type",    StringType(),  nullable=True),
    StructField("country",         StringType(),  nullable=True),
])

customer_df = spark.read.schema(customer_schema).csv(
    "abfss://raw@retaildatalake.dfs.core.windows.net/customers/customers.csv",
    header=True
)

customer_df.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("country") \
    .option("overwriteSchema", "true") \
    .saveAsTable("retailcatalog.sales.retail_customers")

5.2 Converting Existing Parquet to Delta

from delta.tables import DeltaTable

# Convert a Parquet directory to a Delta Table
# (in-place, no data copy needed!)
DeltaTable.convertToDelta(
    spark,
    "parquet.`abfss://raw@retaildatalake.dfs.core.windows.net/customers/customers.parquet`"
)

# Or via SQL
spark.sql("""
    CONVERT TO DELTA parquet.`abfss://raw@retaildatalake.dfs.core.windows.net/customers/customers.parquet`
""")

# For partitioned tables, specify the partition schema
DeltaTable.convertToDelta(
    spark,
    "parquet.`abfss://raw@retaildatalake.dfs.core.windows.net/customers/`",
    "country STRING"  # Partition column schema
)

print("Parquet → Delta conversion complete!")

6. Loading Data from ADLS Gen2

6.1 Batch Load with COPY INTO

-- COPY INTO: Idempotent and incremental
-- Loads only new files on each run
COPY INTO retailcatalog.sales.online_orders
FROM (
    SELECT
        OrderId           AS order_id,
        CustomerId        AS customer_id,
        CAST(OrderDate AS DATE) AS order_date,
        ProductTitle      AS product_title,
        Quantity          AS quantity,
        CAST(TotalAmount AS DOUBLE) AS total_amount,
        'ONLINE'          AS order_channel
    FROM 'abfss://csv-data@retailstorage.dfs.core.windows.net/orders/'
)
FILEFORMAT = CSV
FORMAT_OPTIONS (
    'header' = 'true',
    'sep' = ',',
    'inferSchema' = 'true',
    'nullValue' = 'NULL'
)
COPY_OPTIONS (
    'mergeSchema' = 'true',
    'force' = 'false'  -- false = idempotent (skip already-loaded files)
);

-- Verify results
SELECT COUNT(*) AS total_orders, MIN(order_date), MAX(order_date)
FROM retailcatalog.sales.online_orders;

6.2 Batch Load from an External Location

# Configure access via Unity Catalog External Location
spark.sql("""
    CREATE STORAGE CREDENTIAL retail_storage_credential
    USING MANAGED IDENTITY
""")

spark.sql("""
    CREATE EXTERNAL LOCATION retail_raw
    URL 'abfss://csv-data@retailstorage.dfs.core.windows.net'
    WITH (STORAGE CREDENTIAL retail_storage_credential)
""")

# Read directly from the External Location
orders_df = spark.read.csv(
    "abfss://csv-data@retailstorage.dfs.core.windows.net/orders/",
    header=True,
    inferSchema=True
)

# Load into the Delta Table
orders_df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("retailcatalog.sales.online_orders")

print(f"Rows loaded: {orders_df.count():,}")

7. Streaming to Delta Tables

7.1 Spark Streaming → Delta Architecture

graph LR
    subgraph "Streaming Sources"
        S3[Amazon S3\nNew files]
        EH[Azure Event Hubs\nMessages]
        Kafka[Apache Kafka\nTopics]
        ADLS2[ADLS Gen2\nFile Arrival]
    end
    
    subgraph "Spark Structured Streaming"
        Stream[readStream\nAuto Loader / Kafka]
        Transform[Transformations\nWindowAgg, Watermark]
        Write[writeStream\nDelta Output]
    end
    
    subgraph "Delta Lake (target)"
        DT[Delta Table\nStreamable\nUnified batch+stream]
        CL[Checkpoint\nLocation]
    end
    
    S3 & EH & Kafka & ADLS2 --> Stream
    Stream --> Transform --> Write
    Write --> DT & CL

7.2 Complete Streaming Code

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

# Configure S3 access (or use ADLS Gen2 for Azure-native)
spark.conf.set(
    "fs.s3a.access.key",
    dbutils.secrets.get(scope="aws-secrets", key="access-key-id")
)
spark.conf.set(
    "fs.s3a.secret.key",
    dbutils.secrets.get(scope="aws-secrets", key="secret-access-key")
)

# Streaming data schema
events_schema = StructType([
    StructField("event_id",    StringType(),  nullable=False),
    StructField("user_id",     IntegerType(), nullable=True),
    StructField("product_id",  StringType(),  nullable=True),
    StructField("quantity",    IntegerType(), nullable=True),
    StructField("unit_price",  DoubleType(),  nullable=True),
    StructField("timestamp",   StringType(),  nullable=True),
])

# Read stream from S3/ADLS
events_stream = (
    spark.readStream
    .format("cloudFiles")  # Auto Loader
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation",
            "abfss://checkpoints@retaildatalake.dfs.core.windows.net/schema/events/")
    .schema(events_schema)
    .load("s3a://retail-streaming-data/events/")
)

# Streaming transformations
events_enriched = (
    events_stream
    .withColumn("total_amount", F.col("quantity") * F.col("unit_price"))
    .withColumn("ingestion_time", F.current_timestamp())
    .withColumn("event_timestamp", F.to_timestamp("timestamp"))
    .filter(F.col("quantity") > 0)
    .filter(F.col("unit_price") > 0)
)

# Write to Delta Table
query = (
    events_enriched.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation",
            "abfss://checkpoints@retaildatalake.dfs.core.windows.net/stream/events/")
    .option("mergeSchema", "true")
    .trigger(processingTime="30 seconds")
    .toTable("retailcatalog.sales.streaming_events")
)

print("Stream started!")
query.awaitTermination()

8. Schema Enforcement and Evolution

8.1 Schema Enforcement — Protection Against Bad Data

from pyspark.sql.types import *

# Strict schema defined
customer_schema = StructType([
    StructField("customer_id",     IntegerType(), nullable=False),
    StructField("account_balance", DoubleType(),  nullable=True),
    StructField("country",         StringType(),  nullable=True),
])

# Create table with this schema
initial_data = spark.createDataFrame(
    [(1, 15000.0, "USA"), (2, 8500.0, "Canada")],
    customer_schema
)
initial_data.write.format("delta").mode("overwrite").saveAsTable("default.customer_demo")

# Attempt to write with incompatible schema → ERROR
bad_data = spark.createDataFrame(
    [(3, "high_balance", "UK")],  # account_balance is String instead of Double!
    ["customer_id", "account_balance", "country"]
)

try:
    bad_data.write.format("delta").mode("append").saveAsTable("default.customer_demo")
    print("❌ Should have failed!")
except Exception as e:
    print(f"✅ Schema Enforcement worked correctly!")
    print(f"   Error: {str(e)[:100]}")

8.2 Schema Evolution — Automatically Adding Columns

# Data with new column "age"
new_data_with_age = spark.createDataFrame(
    [(3, 25000.0, "UK", 35), (4, 42000.0, "France", 28)],
    ["customer_id", "account_balance", "country", "age"]  # "age" is new!
)

# WITHOUT mergeSchema → Error
try:
    new_data_with_age.write.format("delta").mode("append").saveAsTable("default.customer_demo")
    print("❌ Should have failed!")
except Exception as e:
    print(f"✅ Expected error without mergeSchema: {str(e)[:80]}")

# WITH mergeSchema = true → Success, schema evolves
new_data_with_age.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("default.customer_demo")

print("\n✅ Schema evolved successfully!")
print("New schema:")
spark.sql("DESCRIBE TABLE default.customer_demo").show()

# Verify: old records have age = NULL
spark.sql("SELECT * FROM default.customer_demo ORDER BY customer_id").show()

8.3 Configuring Delta Lake Constraints

-- NOT NULL constraint
ALTER TABLE retailcatalog.sales.retail_customers
ADD CONSTRAINT customer_id_not_null CHECK (customer_id IS NOT NULL);

-- CHECK constraint (value validation)
ALTER TABLE retailcatalog.sales.retail_customers
ADD CONSTRAINT positive_balance CHECK (account_balance >= 0);

-- Domain constraint
ALTER TABLE retailcatalog.sales.retail_customers
ADD CONSTRAINT valid_country CHECK (country IN ('USA', 'Canada', 'UK', 'France', 'Germany'));

-- List constraints
SHOW CONSTRAINTS IN retailcatalog.sales.retail_customers;

-- Drop a constraint
ALTER TABLE retailcatalog.sales.retail_customers
DROP CONSTRAINT valid_country;

9. Delta Lake Optimization Techniques

9.1 Overview of Optimization Techniques

graph TB
    OPT[Delta Lake\nOptimizations] --> Cache[Caching\nLocal data]
    OPT --> Skip[Data Skipping\nIgnore files]
    OPT --> Layout[Layout Ops\nReorganize data]
    
    Cache --> DeltaCache[Delta Cache\nHardware SSDs]
    Cache --> SparkCache[Spark Cache\ncache / persist]
    
    Skip --> Stats[Column Statistics\nMin, Max, Count]
    Skip --> BloomFilter[Bloom Filters\nhigh-cardinality]
    
    Layout --> Part[Partitioning\nDirectories]
    Layout --> ZO[Z-Order\nCo-location]
    Layout --> Compact[OPTIMIZE\nCompaction]
    Layout --> LC[Liquid Clustering\nAdaptive]

10. Partitioning — Directory-based Organization

10.1 When and How to Partition

graph LR
    subgraph "Without Partitioning"
        T1[Table: 50M rows\nin bulk] --> Q1["SELECT * WHERE country='USA'"]
        Q1 --> SCAN[Scan ALL files\n50M rows read]
    end
    
    subgraph "With Partitioning by country"
        T2[Table: 50M rows] --> P1[country=USA/\n20M rows]
        T2 --> P2[country=Canada/\n15M rows]
        T2 --> P3[country=UK/\n15M rows]
        Q2["SELECT * WHERE country='USA'"] --> P1
        P1 --> SCAN2[Only 20M rows\nPartition Pruning ✅]
    end

10.2 Partitioning Best Practices

# Optimal partitioning for sales data
sales_enriched.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("SaleYear", "SaleMonth") \
    .saveAsTable("retailcatalog.sales.transactions")

# Verify partition structure
spark.sql("""
    SELECT SaleYear, SaleMonth, COUNT(*) AS num_rows
    FROM retailcatalog.sales.transactions
    GROUP BY SaleYear, SaleMonth
    ORDER BY SaleYear, SaleMonth
""").show()

# Partition-optimized query
result = spark.sql("""
    SELECT *
    FROM retailcatalog.sales.transactions
    WHERE SaleYear = 2023
      AND SaleMonth = 12
""")
# → Spark reads ONLY the 2023/12 partition, not the other 50!

11. Z-Order — Data Co-location

11.1 How Z-Order Works

graph LR
    subgraph "Before Z-Order"
        F1[File 1\nQuantity: 1-50\nCountry: mixed] 
        F2[File 2\nQuantity: 25-75\nCountry: mixed]
        F3[File 3\nQuantity: 45-100\nCountry: mixed]
        Q["SELECT WHERE quantity=35"] --> F1
        Q --> F2
        Q --> F3
        Note1[❌ 3 files read]
    end
    
    subgraph "After Z-Order BY quantity"
        ZF1[File 1\nQuantity: 1-30\nData co-located]
        ZF2[File 2\nQuantity: 31-65\nData co-located]
        ZF3[File 3\nQuantity: 66-100\nData co-located]
        ZQ["SELECT WHERE quantity=35"] --> ZF2
        Note2[✅ Only 1 file read\nFiles 1 and 3 skipped]
    end

11.2 Applying Z-Order and Measuring Impact

-- Measure performance BEFORE Z-Order
SELECT COUNT(*) FROM samples.tpch.lineitem
WHERE l_quantity = 35;
-- → Result: ~601,000 rows, duration: 1.58s (full scan 29M rows)

-- Copy the table to your workspace
CREATE OR REPLACE TABLE default.lineitem_zorder
AS SELECT * FROM samples.tpch.lineitem;

SELECT COUNT(*) FROM default.lineitem_zorder;
-- → 29,000,000 rows

-- Apply OPTIMIZE with Z-Order on the frequent filter column
OPTIMIZE default.lineitem_zorder
ZORDER BY (l_quantity);
-- → Compacts AND reorganizes data by l_quantity

-- Measure performance AFTER Z-Order
SELECT COUNT(*) FROM default.lineitem_zorder
WHERE l_quantity = 35;
-- → Same result: ~601,000 rows, duration: 0.45s (3.5x faster!)
-- Reason: Data Skipping ignores files without l_quantity=35

# Z-Order via Python
from delta.tables import DeltaTable

delta_table = DeltaTable.forName(spark, "default.lineitem_zorder")

# Optimize with Z-Order on multiple columns
# (Use max 4 columns for effective Z-Order)
delta_table.optimize() \
    .where("l_shipdate >= '2023-01-01'") \
    .executeZOrderBy("l_quantity", "l_discount")

print("Z-Order applied successfully!")

# Verify Delta stats
spark.sql("""
    DESCRIBE DETAIL default.lineitem_zorder
""").select("numFiles", "sizeInBytes", "numRows").show()

11.3 Ideal Columns for Z-Order

12. OPTIMIZE — Small File Compaction

12.1 The Small Files Problem

graph TB
    subgraph "Before OPTIMIZE (small files)"
        W1[Worker 1] --> F1[f1: 2MB]
        W2[Worker 2] --> F2[f2: 1.5MB]
        W3[Worker 3] --> F3[f3: 3MB]
        W4[Worker 4] --> F4[f4: 1MB]
        W5[Worker 5] --> F5[f5: 2.5MB]
        
        Overhead[❌ Overhead: opening 200 files\nSlow scan, excessive metadata]
    end
    
    subgraph "After OPTIMIZE (bin packing)"
        BIG1[f201: 512MB\nOptimized file]
        BIG2[f202: 488MB\nOptimized file]
        Fast[✅ Only 2 files\n10x faster reads]
    end

12.2 OPTIMIZE Commands

-- Simple OPTIMIZE (bin packing — 1 GB target size)
OPTIMIZE default.lineitem_zorder;

-- OPTIMIZE on a specific partition (recommended for large tables)
OPTIMIZE retailcatalog.sales.transactions
WHERE SaleYear = 2023 AND SaleMonth = 12;

-- OPTIMIZE with simultaneous Z-Order
OPTIMIZE retailcatalog.sales.transactions
ZORDER BY (StoreId, RegionId);

-- OPTIMIZE with Z-Order on a partition
OPTIMIZE retailcatalog.sales.transactions
WHERE SaleYear = 2023
ZORDER BY (StoreId, RegionId);

# Automate OPTIMIZE regularly
from delta.tables import DeltaTable
import datetime

def run_table_maintenance(table_name: str, zorder_cols: list = None):
    """Complete maintenance of a Delta table."""
    
    dt = DeltaTable.forName(spark, table_name)
    
    print(f"=== Maintenance: {table_name} ===")
    
    # 1. OPTIMIZE
    if zorder_cols:
        print(f"OPTIMIZE with Z-Order on: {zorder_cols}")
        dt.optimize().executeZOrderBy(*zorder_cols)
    else:
        print("OPTIMIZE (bin packing)")
        dt.optimize().executeCompaction()
    
    # 2. VACUUM (clean up old files)
    print("VACUUM (7-day retention)")
    dt.vacuum(retentionHours=168)
    
    # 3. Post-maintenance stats
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    print(f"  Files: {detail['numFiles']}")
    print(f"  Size: {round(detail['sizeInBytes'] / (1024**3), 2)} GB")
    print(f"  Avg size: {round(detail['sizeInBytes'] / detail['numFiles'] / (1024**2), 1)} MB/file")
    
    return detail

# Run maintenance
run_table_maintenance(
    "retailcatalog.sales.transactions",
    zorder_cols=["StoreId", "RegionId"]
)

13. Liquid Clustering — Adaptive Partitioning

13.1 Why Liquid Clustering?

Problems with classic partitioning:

• Must be defined at table creation (immutable)
• Can only filter on partition columns
• Creates too many small files with high cardinality

Liquid Clustering: Flexible and evolving partitioning

-- Create a table with Liquid Clustering
CREATE OR REPLACE TABLE retailcatalog.sales.transactions_lc
CLUSTER BY (StoreId, RegionId, SaleYear, SaleMonth)
AS SELECT * FROM retailcatalog.sales.transactions;

-- Modify clustering columns without recreating the table (major advantage!)
ALTER TABLE retailcatalog.sales.transactions_lc
CLUSTER BY (StoreId, ProductCategory, SaleYear);

-- Apply clustering (via OPTIMIZE)
OPTIMIZE retailcatalog.sales.transactions_lc;

-- Liquid Clustering improves progressively with each OPTIMIZE
-- Unlike classic partitioning which is fixed

13.2 Layout Technique Comparison

14. Delta Caching — Local Acceleration

14.1 Types of Caching in Databricks

14.2 Nodes with Delta Cache Accelerated

# Azure node types with Delta Cache Accelerated
# (format: Standard_Ddds_v5, Standard_E8ds_v5, etc.)
# These nodes have fast NVMe SSDs for Delta Cache

# In cluster configuration:
{
  "node_type_id": "Standard_D4ds_v5",  # Delta Cache Accelerated ✅
  "photon_enabled": true  # Photon recommended with Delta Cache
}

14.3 Explicit Caching with Spark

# Cache a frequently accessed table
spark.sql("CACHE TABLE retailcatalog.sales.store_zones")
spark.sql("CACHE TABLE retailcatalog.dimensions.rate_codes")

# Verify cache
spark.catalog.isCached("retailcatalog.sales.store_zones")  # → True

# Cache a DataFrame
popular_stores_df = spark.sql("""
    SELECT StoreId, COUNT(*) AS num_transactions
    FROM retailcatalog.sales.transactions
    GROUP BY StoreId
    ORDER BY num_transactions DESC
    LIMIT 100
""")

popular_stores_df.cache()  # Cache the result of this aggregation

# First run populates the cache
first_run = popular_stores_df.count()  # Slow (actual computation)
second_run = popular_stores_df.count()  # Fast (from cache)

# Release cache
popular_stores_df.unpersist()
spark.sql("UNCACHE TABLE retailcatalog.sales.store_zones")

15. Data Skipping and Column Statistics

15.1 How Data Skipping Works

Delta Lake automatically maintains per-file Parquet statistics:

graph LR
    DL[Delta Lake] --> Stats["Column Statistics\nper Parquet file"]
    Stats --> Min["Min value\n(e.g., min_quantity=1)"]
    Stats --> Max["Max value\n(e.g., max_quantity=30)"]
    Stats --> Count["Null count\n(e.g., null_count=0)"]
    
    Query["SELECT WHERE quantity=35"] --> Engine[Query Engine]
    Engine --> Check{Check stats\nfor each file}
    Check -->|quantity range 1-30\n35 NOT IN 1-30| Skip[SKIP this file ✅]
    Check -->|quantity range 31-65\n35 IN 31-65| Read[Read this file]

15.2 Column Statistics — JSON Structure

// Statistics stored in the DeltaLog for each file
{
  "stats": {
    "numRecords": 75000,
    "minValues": {
      "customer_id": 1001,
      "account_balance": 0.0,
      "open_date": "2010-01-15"
    },
    "maxValues": {
      "customer_id": 76000,
      "account_balance": 250000.0,
      "open_date": "2024-12-31"
    },
    "nullCount": {
      "account_balance": 150,
      "open_date": 0
    }
  }
}

15.3 Bloom Filters for High-Cardinality Columns

-- Enable Bloom Filters for a high-cardinality column
ALTER TABLE retailcatalog.sales.transactions
SET TBLPROPERTIES (
    'delta.dataSkippingNumIndexedCols' = '32',
    'delta.bloomFilter.columns' = 'customer_id',
    'delta.bloomFilter.customer_id.fpp' = '0.1',    -- 10% false positives
    'delta.bloomFilter.customer_id.numItems' = '10000000'
);

-- Example query that benefits from Bloom Filter
SELECT * FROM retailcatalog.sales.transactions
WHERE customer_id = '8f4a7b2c-1234-5678-abcd-ef0123456789';
-- Without Bloom Filter: scan ALL files
-- With Bloom Filter: 90% of files are immediately skipped

16. Photon Acceleration

16.1 What is Photon?

Photon is Databricks’ native vectorized query engine, written in C++:

16.2 Measuring Photon Impact

# Benchmark Photon ON vs OFF
import time

large_query = """
    SELECT 
        l_returnflag,
        l_linestatus,
        SUM(l_quantity) AS sum_qty,
        SUM(l_extendedprice) AS sum_base_price,
        SUM(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
        SUM(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
        AVG(l_quantity) AS avg_qty,
        AVG(l_extendedprice) AS avg_price,
        AVG(l_discount) AS avg_disc,
        COUNT(*) AS count_order
    FROM default.lineitem_zorder
    GROUP BY l_returnflag, l_linestatus
    ORDER BY l_returnflag, l_linestatus
"""

# Test with Photon (on cluster with Photon enabled)
start = time.time()
spark.sql(large_query).collect()
photon_time = time.time() - start

print(f"With Photon:    {photon_time:.2f}s")
print("\nNote: On a cluster without Photon, the same query takes 3–8x longer!")

# Check if Photon is active
photon_enabled = spark.conf.get("spark.databricks.photon.enabled", "false")
print(f"\nPhoton enabled: {photon_enabled}")

16.3 Operations That Benefit Most from Photon

17. Time Travel and Versioning

17.1 Querying Previous Versions

-- View the complete table history
DESCRIBE HISTORY retailcatalog.sales.online_orders;

-- Time Travel by version number
SELECT * FROM retailcatalog.sales.online_orders VERSION AS OF 0;
SELECT * FROM retailcatalog.sales.online_orders VERSION AS OF 3;

-- Time Travel by timestamp
SELECT * FROM retailcatalog.sales.online_orders
TIMESTAMP AS OF '2024-01-15 10:00:00';

SELECT * FROM retailcatalog.sales.online_orders
TIMESTAMP AS OF (current_timestamp() - INTERVAL 1 DAY);

-- Compare two versions
SELECT 
    'Current version' AS source,
    COUNT(*) AS num_rows
FROM retailcatalog.sales.online_orders

UNION ALL

SELECT 
    'Version 0 (initial)',
    COUNT(*)
FROM retailcatalog.sales.online_orders VERSION AS OF 0;

17.2 Restoring a Previous Version (RESTORE)

-- Restore table to a specific version
RESTORE TABLE retailcatalog.sales.online_orders TO VERSION AS OF 2;

-- Or by timestamp
RESTORE TABLE retailcatalog.sales.online_orders 
TO TIMESTAMP AS OF '2024-01-14 12:00:00';

-- Verify the restore
SELECT COUNT(*) FROM retailcatalog.sales.online_orders;
DESCRIBE HISTORY retailcatalog.sales.online_orders;

from delta.tables import DeltaTable

# Restore via Python
dt = DeltaTable.forName(spark, "retailcatalog.sales.online_orders")

# Restore to version 2
dt.restoreToVersion(2)
print("Table restored to version 2!")

# Or by timestamp
from datetime import datetime, timedelta
target_time = datetime.now() - timedelta(hours=24)
dt.restoreToTimestamp(target_time.isoformat())
print(f"Table restored to timestamp: {target_time}")

18. VACUUM — Cleaning Up Obsolete Files

18.1 Why VACUUM Is Necessary

graph TB
    subgraph "Without VACUUM (accumulation)"
        OP0[Version 0: f1.parquet, f2.parquet]
        OP1[Version 1: f3.parquet added] --> OP0
        OP2[Version 2: f1 removed, f4 added] --> OP1
        OP3[Version N: ...] --> OP2
        
        Files[f1.parquet ← No longer referenced but STILL PRESENT\nf2.parquet, f3.parquet, f4.parquet...]
        Cost[Storage cost growing without limit!]
    end
    
    subgraph "With VACUUM RETAIN 168 HOURS"
        VClean[VACUUM deletes f1.parquet\nnot referenced for > 7 days]
        SpaceSaved[Optimized storage ✅\nTime Travel still possible for 7 days]
    end

18.2 Running VACUUM

-- Preview files that would be deleted (DRY RUN)
VACUUM retailcatalog.sales.online_orders DRY RUN;

-- Delete files > 7 days (default = 168 hours)
VACUUM retailcatalog.sales.online_orders;

-- Custom retention (30 days)
VACUUM retailcatalog.sales.online_orders RETAIN 720 HOURS;

-- ⚠️ DANGER: Disable the protection (DO NOT do this in production!)
-- SET spark.databricks.delta.retentionDurationCheck.enabled = false;
-- VACUUM retailcatalog.sales.online_orders RETAIN 0 HOURS;
-- ☠️ Would completely destroy Time Travel!

from delta.tables import DeltaTable

dt = DeltaTable.forName(spark, "retailcatalog.sales.online_orders")

# Dry run before the actual VACUUM
print("=== Dry Run VACUUM ===")
dt.vacuum(0)  # Shows what would be deleted

# Apply VACUUM with 7-day retention
print("\n=== Actual VACUUM (7 days) ===")
dt.vacuum(168)  # 168 hours = 7 days
print("VACUUM complete!")

19. Monitoring Delta Tables

19.1 Databricks Lakehouse Monitoring

# Configure a monitor on a Delta table
# Via the Databricks Lakehouse Monitoring API

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import MonitorCronSchedule, MonitorTimeSeries

w = WorkspaceClient()

# Create a monitor for a data table
monitor = w.quality_monitors.create(
    table_name="retailcatalog.sales.transactions",
    
    # Analysis type
    time_series_profile=MonitorTimeSeries(
        timestamp_col="SaleTime",
        granularities=["1 day", "1 week"]
    ),
    
    # Notifications
    notifications={
        "on_new_classification_tag_detected": {
            "email_addresses": ["data-team@company.com"]
        }
    },
    
    # Refresh schedule
    schedule=MonitorCronSchedule(
        quartz_cron_expression="0 0 8 * * ?",  # Daily at 8am
        timezone_id="America/New_York"
    ),
    
    # Where to store metrics
    output_schema_name="retailcatalog.monitoring"
)

print(f"Monitor created for: {monitor.table_name}")

19.2 Manual Monitoring Queries

-- Analyze the health of a Delta table
SELECT
    version,
    timestamp,
    operation,
    operationParameters,
    operationMetrics.numOutputRows,
    operationMetrics.numFiles,
    userMetadata
FROM (DESCRIBE HISTORY retailcatalog.sales.transactions)
ORDER BY version DESC
LIMIT 20;

-- Detect anomalies: sudden increase in file count
SELECT 
    version,
    timestamp,
    operationMetrics.numFiles AS files_added,
    LAG(operationMetrics.numFiles) OVER (ORDER BY version) AS prev_files
FROM (DESCRIBE HISTORY retailcatalog.sales.transactions)
WHERE operationMetrics.numFiles IS NOT NULL
HAVING files_added > prev_files * 2;  -- Doubling of file count = anomaly

20. Optimistic Concurrency Control

20.1 Conflict Scenarios and Solutions

import time
import random
from delta.tables import DeltaTable

def update_with_retry(table_name: str, condition: str, set_expr: dict, 
                       max_retries: int = 3):
    """
    Update a Delta table with retry on conflict.
    Implements the recommended OCC pattern.
    """
    dt = DeltaTable.forName(spark, table_name)
    
    for attempt in range(max_retries):
        try:
            dt.update(condition=condition, set=set_expr)
            print(f"✅ Update successful (attempt {attempt + 1})")
            return True
        except Exception as e:
            if "concurrent" in str(e).lower() or "newer version" in str(e).lower():
                wait_time = 2 ** attempt + random.random()  # Exponential backoff
                print(f"⚠️  Conflict detected, retrying in {wait_time:.1f}s "
                      f"(attempt {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise  # Different error, don't retry
    
    print(f"❌ Failed after {max_retries} attempts")
    return False

# Usage
success = update_with_retry(
    table_name="retailcatalog.sales.retail_customers",
    condition="account_balance < 1000",
    set_expr={"account_balance": "account_balance + 100"}
)

21. Advanced Azure Use Cases

21.1 Integration with Azure Synapse Analytics

# Read a Delta Table from Azure Synapse (Serverless SQL)
# Synapse can read directly from Parquet+DeltaLog files

# In Azure Synapse (SQL script)
synapse_query = """
-- Read a Delta table from ADLS Gen2
SELECT TOP 100 *
FROM OPENROWSET(
    BULK 'https://retaildatalake.dfs.core.windows.net/tables/transactions/',
    FORMAT = 'DELTA'
) AS t
WHERE t.SaleYear = 2023;
"""

# OR via external view in Synapse
create_view_sql = """
CREATE OR REPLACE VIEW dbo.transactions_view AS
SELECT *
FROM OPENROWSET(
    BULK 'https://retaildatalake.dfs.core.windows.net/tables/transactions/',
    FORMAT = 'DELTA'
) AS t;
"""

21.2 Power BI from Delta Lake

# Expose Delta Tables via Databricks SQL Warehouse
# Power BI connects via the Azure Databricks connector

# Configuration in Power BI Desktop:
# 1. Get Data → Azure → Azure Databricks
# 2. Server: adb-xxxx.azuredatabricks.net
# 3. HTTP Path: /sql/1.0/warehouses/xxxxx
# 4. Authentication: Personal Access Token or AAD

# Optimization for Power BI — pre-compute aggregates
spark.sql("""
    CREATE OR REPLACE TABLE retailcatalog.gold.daily_revenue AS
    SELECT 
        SaleYear,
        SaleMonth,
        SaleDay,
        StoreId,
        COUNT(*) AS num_transactions,
        SUM(TotalAmount) AS total_revenue,
        AVG(TipPercentage) AS avg_tip_pct,
        AVG(TransactionAmount) AS avg_transaction
    FROM retailcatalog.sales.transactions
    GROUP BY SaleYear, SaleMonth, SaleDay, StoreId
""")

print("Power BI view created in the Gold layer!")

22. Summary and Best Practices

22.1 Delta Lake Optimization Checklist

mindmap
  root((Delta Lake\nOptimization))
    Layout
      Partition by frequent filter columns
      Z-Order on analytical columns
      Regular OPTIMIZE (weekly)
      Liquid Clustering for new tables
    Performance
      Photon enabled in production
      Delta Cache Accelerated nodes
      AQE always enabled
      Broadcast join for small tables
    Maintenance
      Weekly VACUUM (7-30 days)
      Analyze DESCRIBE DETAIL
      Monitor avg file size
      Checkpoint every 10 versions
    Quality
      Schema Enforcement always on
      Explicit mergeSchema on evolution
      NOT NULL and CHECK constraints
      Time Travel configured as needed
    Costs
      Z-Order reduces scans
      VACUUM frees storage
      ZSTD compression for archiving
      Avoid micro-partitions

22.2 Decision Table: Which Technique to Use?

23. Glossary

Module 2 – Working with Delta Lake

The Two-Tier Architecture Problem

Before (Two-Tier Architecture):
Data Lake (raw, no schema, low cost)
    ↓ ETL pipelines
Data Warehouse (structured, fast, expensive)

With Delta Lakehouse:
Data Lake (ADLS Gen2)
    + Delta Lake (metadata, transactions, indexes, caching)
    = Lakehouse (the best of both worlds)

Storage Formats Compared

Physical Structure of a Delta Table

/customer_table/
  ├── _delta_log/                  ← Transaction Log
  │     ├── 00000000000000000000.json  (commit 0: creation)
  │     ├── 00000000000000000001.json  (commit 1: insert)
  │     ├── 00000000000000000002.json  (commit 2: update)
  │     └── 00000000000000000010.checkpoint.parquet (optimized snapshot)
  ├── file-01.parquet
  ├── file-02.parquet
  └── file-03.parquet

• Parquet files are NEVER modified in place.
• Modifications create new files.
• The DeltaLog references which files are “active”.

ACID Properties

Creating and Reading Delta Tables

-- Create a Delta Table via SQL
CREATE OR REPLACE TABLE retail_orders (
    order_id BIGINT,
    product_title STRING,
    quantity INT,
    price DOUBLE,
    order_date DATE
) USING DELTA;

-- Load a batch from ADLS
COPY INTO retail_orders
FROM 'abfss://csv-data@retailstorage.dfs.core.windows.net/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true');

# PySpark: batch load
df = spark.read.csv(
    "abfss://csv-data@retailstorage.dfs.core.windows.net/retail_orders.csv",
    header=True
)
df.write.format("delta").saveAsTable("default.retail_orders")

Batch Load from ADLS Gen2

1. Create a Storage Credential → Databricks Access Connector (Managed Identity).
2. Create an External Location → point to the ADLS container.
3. Use COPY INTO or spark.read from the path.

Convert Parquet → Delta Table

-- Convert existing Parquet files to Delta
CONVERT TO DELTA parquet.`abfss://parquet-data@retailstorage.dfs.core.windows.net/customers/`;

Streaming to Delta Tables

# Read from Amazon S3 (or EventHub) in streaming
df_stream = spark.readStream \
    .format("kinesis") \
    .option("streamName", "mystream") \
    .load()

# Write to Delta with checkpointing
df_stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/delta/events/_checkpoints/etl-from-kinesis") \
    .start("abfss://delta@retailstorage.dfs.core.windows.net/events/")

Schema Evolution

# Initial schema → 2 columns (id, name)
df_v1.write.format("delta").saveAsTable("people")

# New schema → 3 columns (id, name, age)
# Error without mergeSchema!
df_v2.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("people")

Module 3 – Optimizing Performance

Optimization Techniques

Partitioning

-- Partition by country and year
CREATE TABLE orders
USING DELTA
PARTITIONED BY (country, year)
AS SELECT * FROM raw_orders;

-- Query that benefits from the partition
SELECT * FROM orders WHERE country = 'USA'; -- reads only country=USA directory

Partitioning best practices:

• ✅ Medium cardinality columns (country, year, month).
• ❌ Avoid high-cardinality columns (customer_id = millions of partitions).
• ❌ Avoid over-partitioning (too many small files = overhead).

Z-Ordering

-- Optimize and co-locate by quantity
OPTIMIZE lineitem_zorder ZORDER BY (l_quantity);

Before Z-Ordering: for SELECT COUNT(*) WHERE l_quantity = 35 → scan all files. After Z-Ordering: records with l_quantity = 35 are co-located → fewer files read.

Caching

# Explicit cache
spark.sql("CACHE SELECT * FROM orders WHERE country = 'USA'")

# Or use Delta Cache Accelerated nodes
# (in Compute → Node type → select "Delta Cache Accelerated")

After first access: data is cached on the Spark node. Subsequent queries: read from cache (much faster).

Photon Acceleration

• Databricks native engine written in C++.
• Compiles SQL/DataFrame operations into optimized machine code.
• Exploits modern CPU features: SIMD (Single Instruction Multiple Data).
• Enabled by default on Databricks clusters.
• Visible in the query plan: PhotonScan, PhotonAggregate.

Measured impact: complex query GROUP BY + ORDER BY + LIMIT on 29M rows:

• Without Photon: ~8 seconds.
• With Photon: ~2 seconds.

Module 4 – Data Integrity and Versioning

Time Travel

-- View the table history
DESCRIBE HISTORY retail_orders;

-- Query a past version
SELECT * FROM retail_orders VERSION AS OF 2;

-- Query by timestamp
SELECT * FROM retail_orders TIMESTAMP AS OF '2024-01-15 10:00:00';

-- Restore to a previous version
RESTORE TABLE retail_orders TO VERSION AS OF 3;

Limits:

• Temporal access up to 7 days back (configurable).
• Transaction log retained 30 days (configurable).
• After VACUUM: old versions are no longer accessible.

VACUUM

-- Preview what would be deleted
VACUUM products DRY RUN;

-- Delete files > 7 days (default)
VACUUM products RETAIN 168 HOURS;

-- Delete files > 0 hours (DANGER: destroys time travel)
SET spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM products RETAIN 0 HOURS;

Monitoring Delta Tables

Catalog Explorer → Table → Quality tab → Get Started

Types of analyses:

Metrics created:

• profile_metrics: descriptive statistics (mean, stddev, null count, etc.).
• drift_metrics: distribution drift over time.

Optimistic Concurrency Control

• Transactions without locks → proceed optimistically.
• Each transaction works on a snapshot of the data.
• At commit time: checks for conflicts with other transactions.
• If conflict detected → rollback and retry.
• Ideal for read-heavy or low-contention workloads.

Azure Use Cases with Delta Lake

Delta Lake Command Reference

Module 5 – Photon Engine

What is Photon?

Photon is Databricks’ native vectorized execution engine, written in C++. It replaces Spark’s classic JVM-based Volcano engine for SQL and DataFrame operations. Photon compiles query plans into optimized machine code leveraging modern CPU features.

flowchart LR
    A[Spark query plan] --> B{Photon\neligible?}
    B -- Yes --> C[C++ compilation\nSIMD / AVX-512]
    B -- No --> D[Classic JVM engine\nstandard Spark]
    C --> E[Vectorized execution\non modern CPU]
    D --> F[Standard execution\nrow-by-row / JVM]
    E --> G[Result]
    F --> G

Internal Architecture

Operations Supported by Photon

Enabling Photon

# Enable via cluster configuration (UI)
# Compute → Edit Cluster → check "Use Photon Acceleration"

# Check if Photon is active on the current cluster
spark.conf.get("spark.databricks.photon.enabled")  # returns "true"

# Temporarily disable for a comparative test
spark.conf.set("spark.databricks.photon.enabled", "false")

-- Check in the query plan
EXPLAIN SELECT l_quantity, COUNT(*) FROM lineitem GROUP BY l_quantity;
-- Look for "PhotonScan", "PhotonAggregate", "PhotonSort" in the plan

Photon vs No-Photon Benchmark

Note: gains vary depending on node type, data cardinality, and plan complexity. Nodes with AVX-512 (e.g., Standard_D4ds_v5) show the best gains.

Recommended Photon Nodes on Azure

Standard_D4ds_v5  → 4 vCores, 16 GB RAM, Delta Cache Accelerated
Standard_D8ds_v5  → 8 vCores, 32 GB RAM, Delta Cache Accelerated
Standard_E4ds_v4  → 4 vCores, 32 GB RAM (high memory)

Module 6 – OPTIMIZE and Z-ORDER

The Small Files Problem

With frequent inserts, streaming, or DML operations (UPDATE, DELETE, MERGE), Delta Lake accumulates many small Parquet files. This drastically degrades read performance because Spark must open and read dozens or hundreds of files for each query.

flowchart TD
    A[100 independent inserts] --> B[100 small files\n~ 1 MB each]
    B --> C{SELECT query}
    C --> D[Spark opens 100 files\nenormous metadata overhead]
    D --> E[Degraded performance\n10× slower]

    F[OPTIMIZE] --> G[Compaction\n100 files → 5 files ~1 GB]
    G --> H{SELECT query}
    H --> I[Spark opens 5 files\nefficient I/O]
    I --> J[Optimal performance]

The OPTIMIZE Command

-- Compact all small files in the table
OPTIMIZE my_delta_table;

-- Compact only a specific partition
OPTIMIZE my_delta_table WHERE country = 'USA';

-- Compact with Z-Ordering (multi-dimensional clustering)
OPTIMIZE my_delta_table ZORDER BY (customer_id, order_date);

# PySpark equivalent
from delta.tables import DeltaTable

delta_table = DeltaTable.forName(spark, "my_delta_table")
delta_table.optimize().executeCompaction()

# With Z-ORDER
delta_table.optimize().executeZOrderBy("customer_id", "order_date")

OPTIMIZE behavior:

• Targets files smaller than 1 GB (configurable target size).
• Creates new consolidated Parquet files.
• Old files become obsolete (not referenced in the log).
• Obsolete files are deleted by VACUUM.
• OPTIMIZE is idempotent: running it again causes no issues.

Z-ORDER BY – Multi-dimensional Clustering

Z-Ordering is a data locality technique that reorganizes records in Parquet files so that similar values for the specified columns are stored in the same or adjacent files.

flowchart LR
    subgraph Before Z-ORDER
        A1[File 1\nQty:12,35,7,50]
        A2[File 2\nQty:1,35,20,8]
        A3[File 3\nQty:35,40,2,35]
    end
    subgraph After Z-ORDER BY qty
        B1[File 1\nQty:1,2,7,8]
        B2[File 2\nQty:12,20,35,35]
        B3[File 3\nQty:35,40,50,...]
    end
    Before --> OPTIMIZE --> After

Principle of the Z-curve: $$Z(x, y) = \text{interleaving of bits of } x \text{ and } y$$

For two columns (x, y), Z-ordering interleaves the bits of their values to create a one-dimensional sort key that preserves spatial locality: points close in 2D space have close Z keys.

-- Typical use case: frequent filters on date AND region
OPTIMIZE sales_table ZORDER BY (sale_date, region);

-- Verify impact: compare files read before/after
-- In Spark UI → SQL → PhotonScan → "files pruned" vs "files read"

Data Skipping Statistics Post-OPTIMIZE

-- View statistics collected per file
SELECT
    path,
    size,
    stats:numRecords,
    stats:minValues,
    stats:maxValues
FROM (
    DESCRIBE DETAIL my_delta_table
);

-- Force statistics collection on the first 32 columns
ALTER TABLE my_delta_table
SET TBLPROPERTIES ('delta.dataSkippingNumIndexedCols' = '32');

Liquid Clustering – The New Approach

Liquid Clustering is the feature introduced in Databricks Runtime 13.3+ that replaces Z-Ordering and traditional partitioning. It uses a more flexible incremental clustering algorithm.

-- Create a table with Liquid Clustering
CREATE TABLE sales_liquid
CLUSTER BY (customer_id, sale_date)
AS SELECT * FROM raw_sales;

-- Modify clustering columns (without full rewrite)
ALTER TABLE sales_liquid CLUSTER BY (region, sale_date);

-- Trigger incremental clustering
OPTIMIZE sales_liquid;

When to Use Z-ORDER vs Liquid Clustering?

Z-ORDER → stable filter columns, DBR < 13.3, existing tables
Liquid Clustering → new tables, changing columns, DBR 13.3+

Recommended OPTIMIZE Frequency

# Example: trigger OPTIMIZE in a daily job
# Databricks Workflow → Task → Notebook or SQL

# Automate via Delta Table Properties
ALTER TABLE orders
SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',   -- merge small files on write
    'delta.autoOptimize.autoCompact'   = 'true'    -- automatic background compaction
);

Module 7 – Delta Lake Internals

The Transaction Log (_delta_log)

The _delta_log is the heart of Delta Lake. Each operation on the table generates a sequentially numbered JSON commit file. These files contain the metadata of all modifications.

/my_table/
  _delta_log/
    00000000000000000000.json   ← commit 0: CREATE TABLE
    00000000000000000001.json   ← commit 1: INSERT
    00000000000000000002.json   ← commit 2: UPDATE
    ...
    00000000000000000009.json   ← commit 9: DELETE
    00000000000000000010.checkpoint.parquet  ← full snapshot (commit 10)
    00000000000000000010.json   ← commit 10
    _last_checkpoint             ← pointer to the last checkpoint

JSON Commit File Structure

{
  "commitInfo": {
    "timestamp": 1700000000000,
    "operation": "WRITE",
    "operationParameters": {"mode": "Append"},
    "isolationLevel": "Serializable",
    "isBlindAppend": true
  },
  "add": {
    "path": "part-00000-abc123.snappy.parquet",
    "size": 1048576,
    "stats": "{\"numRecords\":5000,\"minValues\":{\"id\":1,\"date\":\"2024-01-01\"},\"maxValues\":{\"id\":5000,\"date\":\"2024-12-31\"},\"nullCount\":{\"id\":0}}"
  },
  "remove": {
    "path": "part-00000-old456.snappy.parquet",
    "deletionTimestamp": 1700000000000,
    "dataChange": true
  }
}

Automatic Checkpointing

sequenceDiagram
    participant W as Writer
    participant L as _delta_log
    participant C as Checkpoint

    W->>L: commit 0 (.json)
    W->>L: commits 1..9 (.json)
    W->>L: commit 10 (.json)
    L->>C: Generate checkpoint.parquet\n(snapshot of complete state)
    Note over C: Contains all active files\nand statistics at commit 10
    W->>L: commits 11..19 (.json)
    W->>L: commit 20 (.json)
    L->>C: New checkpoint.parquet

Why checkpoints?

• Without checkpoint, reading the table state → replay ALL commits from the beginning.
• With checkpoint (every 10 commits) → replay only from the last checkpoint.
• Configurable: delta.checkpointInterval (default: 10).

-- Force a manual checkpoint
FSCK REPAIR TABLE my_table;

-- Modify the checkpoint interval
ALTER TABLE my_table
SET TBLPROPERTIES ('delta.checkpointInterval' = '5');

Data Skipping and Min/Max Statistics

Delta Lake automatically collects per-Parquet-file statistics during writes:

Query: SELECT * WHERE order_date = '2024-06-15'

File 1: minDate=2024-01-01, maxDate=2024-03-31 → SKIP ✓
File 2: minDate=2024-04-01, maxDate=2024-06-30 → READ ✓
File 3: minDate=2024-07-01, maxDate=2024-12-31 → SKIP ✓

Result: only 1 file out of 3 is read → 66% files eliminated

Bloom Filter Indexes

Bloom filters complement min/max statistics for high-cardinality columns (e.g., IDs, UUIDs) where min/max are poorly discriminating.

-- Create a bloom filter index on a column
CREATE BLOOMFILTER INDEX ON TABLE orders
FOR COLUMNS (order_id OPTIONS (fpp=0.1, numItems=10000000));

-- Check existing indexes
SHOW TBLPROPERTIES orders;
-- Look for "delta.bloomFilterColumns"

-- Remove a bloom filter
DROP BLOOMFILTER INDEX ON TABLE orders FOR COLUMNS (order_id);

Module 8 – Caching in Depth

Cache Layer Architecture

flowchart TD
    A[Spark Query] --> B{Spark Cache\nin-memory?}
    B -- Hit --> C[Driver/executor RAM\nvery fast]
    B -- Miss --> D{Delta Cache\nNVMe SSD?}
    D -- Hit --> E[Local NVMe SSD\nfast]
    D -- Miss --> F[ADLS Gen2\nremote storage\nslow]
    F --> G[Data read\nand cached to SSD]
    G --> D

Delta Cache (IO Cache)

The Delta Cache is an I/O-level cache layer managed by Databricks. It stores decompressed Parquet data blocks on local NVMe SSD disks of compute nodes.

# Pre-load the cache explicitly (avoids cache miss on first query)
spark.sql("CACHE SELECT * FROM lineitem WHERE l_quantity > 10")

# Check cache state (Spark UI → Storage tab)
# or via SQL
spark.catalog.isCached("lineitem")

# Invalidate table cache
spark.catalog.uncacheTable("lineitem")
spark.sql("UNCACHE TABLE lineitem")

Delta Cache Accelerated Nodes on Azure

Standard_D4ds_v5  → 150 GB NVMe SSD cache
Standard_D8ds_v5  → 300 GB NVMe SSD cache
Standard_L8s_v3   → 1.92 TB NVMe SSD cache (very heavy workloads)

Spark In-Memory Cache (persist/cache)

Unlike the Delta Cache, the Spark in-memory cache stores data in executor RAM. It is managed via the DataFrame API.

from pyspark.storagelevel import StorageLevel

# Simple memory cache (default)
df = spark.table("orders")
df.cache()                            # MEMORY_AND_DISK by default

# Cache with explicit storage level
df.persist(StorageLevel.MEMORY_ONLY)          # RAM only, drop if full
df.persist(StorageLevel.MEMORY_AND_DISK)      # RAM, then spills to disk
df.persist(StorageLevel.DISK_ONLY)            # Disk only
df.persist(StorageLevel.MEMORY_AND_DISK_SER)  # Serialized RAM (less memory)

# Force computation and fill the cache
df.count()  # triggering action

# Release memory
df.unpersist()

-- Cache via SQL
CACHE TABLE orders;
CACHE LAZY TABLE orders;  -- cache only on first access
UNCACHE TABLE orders;

Module 9 – Adaptive Query Execution (AQE)

What is AQE?

Adaptive Query Execution (AQE), introduced in Spark 3.0, is a dynamic query plan optimization mechanism that adapts to actual statistics collected during execution, rather than relying solely on initial planning statistics.

flowchart LR
    A[Initial plan\nestimated statistics] --> B[Stage 1\nexecution]
    B --> C[Collect actual\npartition statistics]
    C --> D{AQE\nre-optimizes}
    D --> E[Adapted plan\nstage 2]
    E --> F[Stage 2\noptimized execution]
    F --> G[Final result]

Enabling AQE

# Enable AQE (enabled by default since Spark 3.2 / DBR 9+)
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Check status
spark.conf.get("spark.sql.adaptive.enabled")

# Disable to compare behavior
spark.conf.set("spark.sql.adaptive.enabled", "false")

1. Dynamic Partition Coalescing

Merges small shuffle partitions produced after a GROUP BY or JOIN to avoid too many lightweight tasks.

# Without AQE: 200 shuffle partitions by default (even if data is small)
spark.conf.set("spark.sql.shuffle.partitions", "200")

# With AQE: dynamic coalesce
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128mb")
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", "1mb")

Example:
- Before AQE: 200 shuffle tasks, 190 of which process < 1 MB of data
- After AQE: coalesced into 10 tasks of ~128 MB each
- Result: 95% fewer tasks, drastically reduced overhead

2. Skew Join Optimization

Detects and handles imbalanced data partitions (skew) in JOINs, which cause “straggler” tasks that slow down the entire job.

# Enable skew detection
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
# A partition is "skewed" if it is 5× larger than the median
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256mb")

3. Dynamic Broadcast Join Conversion

AQE can dynamically convert a sort-merge join into a broadcast join if, after reading the table, a partition turns out to be small enough to fit in memory.

# Threshold for automatic conversion to broadcast join
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "10mb")

AQE Configuration Summary

spark.conf.set("spark.sql.adaptive.enabled",                              "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled",           "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled",                     "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes",         "128mb")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold",           "10mb")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor",       "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256mb")

Module 10 – Join Optimizations

Spark Join Types

flowchart TD
    A[JOIN request] --> B{Size of\nthe small table?}
    B -- "< broadcast threshold\n(10 MB default)" --> C[Broadcast Join\nHashJoin in memory]
    B -- "> threshold, sorted data" --> D[Sort-Merge Join\nmost common]
    B -- "Bucketed tables\nsame bucketing" --> E[Bucket Join\nno shuffle]
    C --> F[Very fast\nno shuffle]
    D --> G[Costly shuffle\nbut scalable]
    E --> H[Very efficient\npre-partitioned]

1. Broadcast Join

The broadcast join distributes the small table to all executors, eliminating shuffle.

from pyspark.sql.functions import broadcast

# Explicit broadcast hint
result = large_orders.join(
    broadcast(small_countries),
    "country_id"
)

# Verify broadcast is used
result.explain()
# Look for "BroadcastHashJoin" or "PhotonBroadcastHashJoin" in the plan

-- SQL hint
SELECT /*+ BROADCAST(c) */ o.*, c.name
FROM orders o
JOIN countries c ON o.country_id = c.id;

# Configure automatic broadcast threshold
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50mb")  # default: 10mb
# Disable automatic broadcast
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

Rule of thumb: use broadcast for tables < 50–100 MB. Beyond that, the broadcast cost outweighs the benefit.

2. Sort-Merge Join

Default algorithm for large tables. Requires a shuffle (data redistribution) followed by sorting on both sides.

# Configure number of shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", "400")
# Rule of thumb: ~2-3 partitions per available CPU core

Search Terms

optimize · storage · performance · delta · lake · azure · databricks · spark · data · engineering · analytics · photon · architecture · tables · z-order · join · cache · caching · clustering · optimization · aqe · evolution · liquid · monitoring