Databricks Data Engineering Mastery

Chapter 1

Chapter 01: PySpark Foundations for Data Engineers

Duration: 6 hours | Difficulty: Intermediate | Prerequisites: Python fundamentals, basic SQL

Learning Objectives

By the end of this module, you will be able to:

1. Configure a SparkSession for local development and Databricks workloads

2. Perform complex transformations using the DataFrame API fluently

3. Write efficient column expressions with pyspark.sql.functions

4. Explain why UDFs are costly and when built-in functions are the right choice

5. Read and interpret Spark execution plans to diagnose performance issues

6. Apply partitioning strategies to control parallelism and data layout

1. SparkSession: Your Entry Point to Spark

Every PySpark program begins with a SparkSession. On Databricks, one is pre-configured as spark, but understanding its internals is essential for testing, tuning, and local development.

1.1 Creating a SparkSession

python

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("DataEngineeringMastery")
    .master("local[*]")  # Use all available cores locally
    .config("spark.sql.shuffle.partitions", "8")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .getOrCreate()
)

1.2 Key Configuration Properties

Property	Default	Recommendation
`spark.sql.shuffle.partitions`	200	Set to 2-3x your cluster cores for small-medium data
`spark.sql.adaptive.enabled`	true (Spark 3.2+)	Always keep enabled
`spark.sql.adaptive.coalescePartitions.enabled`	true	Reduces small partitions after shuffle
`spark.sql.files.maxPartitionBytes`	128 MB	Increase for wide tables, decrease for narrow
`spark.sql.autoBroadcastJoinThreshold`	10 MB	Increase cautiously for dimension tables

1.3 SparkSession on Databricks

On Databricks, the session is pre-initialized. You extend it, not replace it:

python

# Databricks -- spark is already available
# Set additional config at runtime
spark.conf.set("spark.sql.shuffle.partitions", "auto")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# Check current config
print(spark.conf.get("spark.sql.adaptive.enabled"))

Tip: Never call SparkSession.builder.getOrCreate() in a Databricks notebook -- it can silently override cluster-level settings.

2. The DataFrame API Deep-Dive

DataFrames are the workhorse of PySpark. They represent distributed, schema-enforced collections of rows.

2.1 Creating DataFrames

python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType

# From explicit schema (preferred for production)
schema = StructType([
    StructField("order_id", StringType(), nullable=False),
    StructField("customer_id", StringType(), nullable=False),
    StructField("product", StringType(), nullable=True),
    StructField("quantity", IntegerType(), nullable=True),
    StructField("unit_price", DoubleType(), nullable=True),
    StructField("order_date", TimestampType(), nullable=True),
])

df = spark.read.schema(schema).json("/mnt/raw/orders/")

# From a list (testing & prototyping)
data = [
    ("ORD-001", "C-100", "Widget A", 5, 12.99),
    ("ORD-002", "C-101", "Widget B", 2, 24.50),
    ("ORD-003", "C-100", "Widget A", 1, 12.99),
]
columns = ["order_id", "customer_id", "product", "quantity", "unit_price"]
df = spark.createDataFrame(data, columns)

2.2 Core Transformations

Transformations are lazy -- they build a logical plan but do not execute until an action triggers computation.

python

from pyspark.sql import functions as F

# Chained transformations
result = (
    df
    .filter(F.col("quantity") > 0)
    .withColumn("total_price", F.col("quantity") * F.col("unit_price"))
    .withColumn("order_date", F.to_date("order_date"))
    .withColumnRenamed("customer_id", "cust_id")
    .select("order_id", "cust_id", "product", "total_price", "order_date")
    .orderBy(F.col("total_price").desc())
)

2.3 Common DataFrame Operations

python

# Aggregations
revenue_by_product = (
    df
    .groupBy("product")
    .agg(
        F.sum(F.col("quantity") * F.col("unit_price")).alias("total_revenue"),
        F.count("order_id").alias("order_count"),
        F.avg("unit_price").alias("avg_price"),
    )
)

# Window functions
from pyspark.sql.window import Window

window_spec = Window.partitionBy("customer_id").orderBy(F.col("order_date").desc())

df_ranked = df.withColumn(
    "purchase_rank", F.row_number().over(window_spec)
)

# Pivot
pivot_df = (
    df
    .groupBy("customer_id")
    .pivot("product")
    .agg(F.sum("quantity"))
    .fillna(0)
)

2.4 Joins

python

customers = spark.read.table("catalog.schema.dim_customers")
orders = spark.read.table("catalog.schema.fact_orders")

# Inner join
enriched = orders.join(customers, orders.customer_id == customers.id, "inner")

# Left anti join -- orders without matching customers (data quality check)
orphan_orders = orders.join(customers, orders.customer_id == customers.id, "left_anti")

# Broadcast join for small dimension tables
from pyspark.sql.functions import broadcast

enriched = orders.join(broadcast(customers), orders.customer_id == customers.id, "inner")

Rule of Thumb: If the smaller side of a join fits in memory (< 1 GB after serialization), broadcast it.

3. Column Expressions

Column expressions are the building blocks of transformations. Mastering them eliminates the need for UDFs in 90% of cases.

3.1 The `F.col()` Pattern

python

from pyspark.sql import functions as F

# Prefer F.col() for clarity and to avoid ambiguity in joins
df.select(F.col("product"), F.col("quantity"))

# String operations
df.withColumn("product_upper", F.upper(F.col("product")))
df.withColumn("product_trimmed", F.trim(F.col("product")))
df.filter(F.col("product").like("%Widget%"))
df.filter(F.col("product").rlike("^Widget\\s[A-Z]$"))

# Null handling
df.withColumn("product_clean", F.coalesce(F.col("product"), F.lit("UNKNOWN")))
df.filter(F.col("product").isNotNull())
df.withColumn("has_product", F.when(F.col("product").isNull(), False).otherwise(True))

3.2 Conditional Logic with `when`/`otherwise`

python

df_tiered = df.withColumn(
    "price_tier",
    F.when(F.col("unit_price") < 10, "budget")
     .when(F.col("unit_price") < 50, "standard")
     .when(F.col("unit_price") < 100, "premium")
     .otherwise("luxury")
)

3.3 Working with Complex Types

python

# Structs
df.select(F.col("address.city"), F.col("address.zip_code"))

# Arrays
df.withColumn("first_tag", F.element_at(F.col("tags"), 1))
df.withColumn("tag_count", F.size(F.col("tags")))
df.select(F.explode(F.col("tags")).alias("tag"))

# Maps
df.withColumn("value", F.col("metadata")["key_name"])
df.select(F.map_keys(F.col("metadata")).alias("all_keys"))

# JSON strings
df.withColumn(
    "parsed",
    F.from_json(F.col("json_payload"), "struct<name:string,age:int>")
)

3.4 Date and Timestamp Functions

python

df = df.withColumn("event_date", F.to_date("event_timestamp"))
df = df.withColumn("year", F.year("event_date"))
df = df.withColumn("month", F.month("event_date"))
df = df.withColumn("day_of_week", F.dayofweek("event_date"))
df = df.withColumn("date_diff", F.datediff(F.current_date(), F.col("event_date")))
df = df.withColumn("next_month", F.add_months(F.col("event_date"), 1))

4. UDFs vs Built-in Functions

4.1 Why UDFs Are Expensive

User-Defined Functions (UDFs) break the Catalyst optimizer's ability to reason about your code:

1. Serialization overhead: Data is serialized from JVM to Python and back for each row

2. No predicate pushdown: The optimizer cannot push filters through a UDF

3. No code generation: Tungsten's whole-stage code generation is bypassed

4. Memory pressure: Python worker processes consume additional executor memory

python

# BAD: Python UDF for something built-in functions handle
from pyspark.sql.functions import udf

@udf(StringType())
def upper_udf(s):
    return s.upper() if s else None

df.withColumn("product_upper", upper_udf("product"))

# GOOD: Built-in function -- 10-100x faster
df.withColumn("product_upper", F.upper(F.col("product")))

4.2 When UDFs Are Acceptable

Sometimes UDFs are the only option:

python

# Complex business logic that cannot be expressed with built-in functions
import re
from pyspark.sql.types import ArrayType, StringType

@udf(ArrayType(StringType()))
def extract_emails(text):
    """Extract all email addresses from a text field."""
    if text is None:
        return []
    pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
    return re.findall(pattern, text)

df.withColumn("emails", extract_emails(F.col("message_body")))

4.3 Pandas UDFs: The Middle Ground

When you must use a UDF, prefer Pandas UDFs (vectorized UDFs) for 3-100x better performance over standard UDFs:

python

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf(DoubleType())
def normalize_score(series: pd.Series) -> pd.Series:
    """Z-score normalization -- operates on batches, not individual rows."""
    return (series - series.mean()) / series.std()

df.withColumn("normalized_score", normalize_score(F.col("raw_score")))

4.4 Decision Framework

Scenario	Use
String manipulation (upper, trim, regex)	Built-in `F.*` functions
Date arithmetic	Built-in `F.*` functions
Conditional logic	`F.when().otherwise()`
Aggregations	Built-in aggregate functions
Complex regex extraction	`F.regexp_extract` or `F.regexp_replace`
External library calls (ML model scoring)	Pandas UDF
Row-level complex business logic	Pandas UDF
Interacting with external services per row	`mapInPandas` or `applyInPandas`

5. Lazy Evaluation and Execution Plans

5.1 Lazy Evaluation

Spark builds a DAG (Directed Acyclic Graph) of transformations but does not execute them until an action is called.

Transformations (lazy): select, filter, join, groupBy, withColumn, orderBy

Actions (trigger execution): show, count, collect, write, save, toPandas

python

# Nothing executes here -- only a plan is built
filtered = df.filter(F.col("quantity") > 0)
enriched = filtered.withColumn("total", F.col("quantity") * F.col("unit_price"))
grouped = enriched.groupBy("product").agg(F.sum("total").alias("revenue"))

# NOW execution happens
grouped.show()  # Action triggers the full DAG

5.2 Reading Execution Plans

python

# Logical plan (what Spark intends to do)
grouped.explain(mode="simple")

# Physical plan with all optimizations applied
grouped.explain(mode="extended")

# Formatted plan (Spark 3.0+) -- most readable
grouped.explain(mode="formatted")

Example output and what to look for:

== Physical Plan ==
*(2) HashAggregate(keys=[product#12], functions=[sum(total#45)])
+- Exchange hashpartitioning(product#12, 200)    <-- SHUFFLE! Check partition count
   +- *(1) HashAggregate(keys=[product#12], functions=[partial_sum(total#45)])
      +- *(1) Project [product#12, (quantity#14 * unit_price#15) AS total#45]
         +- *(1) Filter (quantity#14 > 0)
            +- *(1) Scan parquet [product#12,quantity#14,unit_price#15]
                     PushedFilters: [IsNotNull(quantity), GreaterThan(quantity,0)]

Key indicators:

Exchange = shuffle (data movement across executors). Minimize these.
BroadcastExchange = broadcast join. Good for small tables.
PushedFilters = filters pushed down to the storage layer. Good.
*(N) = whole-stage code generation stage number. More stages may mean more shuffles.

5.3 The Catalyst Optimizer

Catalyst performs these optimization phases:

1. Analysis: Resolves column names and types against the catalog

2. Logical Optimization: Predicate pushdown, constant folding, column pruning

3. Physical Planning: Selects join strategies, sort vs hash aggregation

4. Code Generation: Tungsten generates JVM bytecode for execution

python

# See the full optimization journey
df.filter(F.col("quantity") > 0).select("product", "quantity").explain(mode="extended")

# Output shows:
# == Parsed Logical Plan ==   (raw user intent)
# == Analyzed Logical Plan == (types resolved)
# == Optimized Logical Plan == (Catalyst optimizations applied)
# == Physical Plan ==          (execution strategy selected)

6. Partitioning Basics

6.1 In-Memory Partitions vs On-Disk Partitions

These are two distinct concepts:

Concept	What It Controls	How to Set
In-memory partitions	Parallelism during computation	`repartition()`, `coalesce()`, config
On-disk partitions	File layout in storage	`partitionBy()` on write

6.2 Controlling In-Memory Partitions

python

# Check current partition count
print(df.rdd.getNumPartitions())  # e.g., 200

# Increase partitions (triggers shuffle)
df_repartitioned = df.repartition(16)

# Decrease partitions (no shuffle -- coalesces in place)
df_coalesced = df.coalesce(4)

# Repartition by column (for downstream join/groupBy optimization)
df_by_date = df.repartition(16, "order_date")

6.3 On-Disk Partitioning

python

# Partition data on disk by date columns
(
    df
    .write
    .format("delta")
    .mode("overwrite")
    .partitionBy("year", "month")
    .save("/mnt/silver/orders")
)

Partitioning guidelines:

Partition by columns used in WHERE clauses (date is the most common)
Each partition should contain at least 128 MB - 1 GB of data
Avoid high-cardinality columns (e.g., user_id with millions of distinct values)
Prefer Delta Lake's liquid clustering over traditional partitionBy for new tables

6.4 The Small File Problem

Over-partitioning creates thousands of tiny files that degrade read performance:

python

# Check file sizes in a Delta table
display(
    spark.sql("""
        DESCRIBE DETAIL catalog.schema.orders
    """)
)

# Compact small files
spark.sql("OPTIMIZE catalog.schema.orders")

# Enable auto-compaction on write
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

Hands-On Exercises

Exercise 1: DataFrame Gymnastics

Using the following dataset, perform the requested transformations:

python

data = [
    ("2026-01-15", "C-001", "laptop", 1, 1299.99, "electronics"),
    ("2026-01-15", "C-002", "mouse", 3, 25.50, "electronics"),
    ("2026-01-16", "C-001", "desk", 1, 499.00, "furniture"),
    ("2026-01-16", "C-003", "keyboard", 2, 75.00, "electronics"),
    ("2026-01-17", "C-002", "monitor", 1, 549.99, "electronics"),
    ("2026-01-17", "C-003", "chair", 1, 329.00, "furniture"),
    ("2026-01-18", "C-001", "webcam", 1, 89.99, "electronics"),
    ("2026-01-18", "C-004", None, 0, 0.00, "unknown"),
]
columns = ["order_date", "customer_id", "product", "quantity", "unit_price", "category"]
orders = spark.createDataFrame(data, columns)

Tasks:

1. Filter out rows where product is null or quantity is 0

2. Add a total_price column (quantity * unit_price)

3. Add a price_tier column: "budget" (< 50), "mid" (50-500), "premium" (> 500)

4. Calculate each customer's cumulative spend ordered by date (window function)

5. Find the top-spending customer per category

Exercise 2: UDF Elimination

Refactor this code to replace UDFs with built-in functions:

python

# Original (slow)
@udf(StringType())
def clean_product_name(name):
    if name is None:
        return "UNKNOWN"
    return name.strip().upper().replace(" ", "_")

@udf(DoubleType())
def apply_discount(price, qty):
    if qty >= 10:
        return price * 0.9
    elif qty >= 5:
        return price * 0.95
    return price

df = df.withColumn("clean_name", clean_product_name("product"))
df = df.withColumn("discounted_price", apply_discount("unit_price", "quantity"))

Challenge: Rewrite both UDFs using only pyspark.sql.functions.

Exercise 3: Execution Plan Analysis

Run the following query and analyze its execution plan:

python

large_orders = orders.filter(F.col("total_price") > 100)
customer_summary = (
    large_orders
    .join(broadcast(customers), "customer_id")
    .groupBy("customer_name", "category")
    .agg(
        F.sum("total_price").alias("total_spend"),
        F.count("*").alias("order_count"),
    )
    .orderBy(F.col("total_spend").desc())
)
customer_summary.explain(mode="formatted")

Questions:

1. How many shuffle stages are there?

2. Is the broadcast join being applied? How can you tell?

3. Where is the filter being applied relative to the join?

4. What would happen if you removed the broadcast() hint?

Exercise 4: Partition Tuning

python

# You have a 10 GB dataset with 200 default shuffle partitions
# The cluster has 8 workers, each with 4 cores (32 total cores)
# The job runs a groupBy followed by a write

# Task 1: What is the ideal shuffle partition count? Why?
# Task 2: Write the code to repartition before writing to Delta
# Task 3: The target table is queried by date range 95% of the time.
#          What partitioning strategy would you use on disk?

Key Takeaways

1. SparkSession is your control plane -- understand its configuration properties for both local and Databricks environments.

2. DataFrames are lazy, distributed, and schema-enforced. Chain transformations fluently and trigger execution only when needed.

3. Column expressions with pyspark.sql.functions cover 90%+ of transformation needs. Master when/otherwise, window functions, and complex type handling.

4. Avoid Python UDFs whenever possible. When unavoidable, use Pandas UDFs for vectorized performance.

5. Read execution plans to understand what Spark actually does. Look for shuffles, pushed filters, and join strategies.

6. Partitioning controls parallelism (in-memory) and data layout (on-disk). Right-size both to avoid the small file problem.

Knowledge Check

1. What is the difference between a transformation and an action in Spark? Give three examples of each.

2. Why does spark.sql.shuffle.partitions = 200 often perform poorly on small datasets?

3. A colleague writes a Python UDF that converts a date string to a timestamp. What do you recommend instead, and why?

4. Explain what Exchange hashpartitioning means in a physical plan.

5. When should you use coalesce() instead of repartition()? What is the key difference?

6. You have a table with 50,000 distinct user_id values. Should you partitionBy("user_id") when writing to Delta? Why or why not?

7. What does the Catalyst optimizer's "predicate pushdown" optimization do, and why does it matter for performance?

Chapter 2

Chapter 02: Delta Lake Deep Dive

Duration: 6 hours | Difficulty: Intermediate-Advanced | Prerequisites: Module 01 (PySpark Foundations)

Learning Objectives

By the end of this module you will be able to:

1. Explain Delta Lake's architecture and how the transaction log ensures ACID guarantees

2. Use time travel for auditing, rollback, and reproducible queries

3. Configure schema enforcement and schema evolution for safe pipeline changes

4. Implement MERGE operations for UPSERT and Slowly Changing Dimension Type 2 patterns

5. Operate Delta tables in production with VACUUM, OPTIMIZE, Z-ORDER, and liquid clustering

6. Enable and consume Change Data Feed for downstream CDC pipelines

1. Delta Lake Architecture

1.1 What Makes Delta Different from Parquet

Delta Lake is an open-source storage layer that adds reliability to data lakes. Under the hood, a Delta table is:

delta_table/
├── _delta_log/              # Transaction log (JSON + Parquet checkpoints)
│   ├── 00000000000000000000.json
│   ├── 00000000000000000001.json
│   ├── 00000000000000000002.json
│   └── 00000000000000000010.checkpoint.parquet
├── part-00000-...-.parquet  # Data files
├── part-00001-...-.parquet
└── part-00002-...-.parquet

The _delta_log/ directory is the brain. Each JSON file is a commit that records:

Files added
Files removed
Metadata changes (schema, partitioning)
Transaction information

1.2 The Transaction Log Protocol

python

# You can inspect the log directly (educational, not for production use)
log_df = spark.read.json("abfss://lake@storage.dfs.core.windows.net/tables/orders/_delta_log/*.json")
log_df.printSchema()

# Each commit contains add/remove actions
adds = log_df.select(F.explode("add").alias("a")).select("a.*")
adds.select("path", "size", "modificationTime", "dataChange").show(truncate=False)

1.3 ACID Transactions

Property	How Delta Achieves It
Atomicity	Writes are committed by appending a single JSON entry to `_delta_log/`. Either the file appears or it doesn't.
Consistency	Schema enforcement rejects writes that don't match the table schema.
Isolation	Optimistic concurrency control. Concurrent writers serialize via conflict detection on the log.
Durability	Data files are immutable Parquet on cloud storage (ADLS, S3). The log is the source of truth.

python

# Concurrent write safety example
# Writer A and Writer B both try to append to the same table simultaneously.
# Delta handles this via optimistic concurrency:
# 1. Both read the latest log version (e.g., version 5)
# 2. Both prepare their commits
# 3. One succeeds writing version 6
# 4. The other detects the conflict, re-reads version 6, and retries as version 7
# This is transparent to the user.

2. Time Travel

Every commit to a Delta table creates a new version. You can query any historical version.

2.1 Querying by Version and Timestamp

python

# By version number
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("/tables/orders")

# By timestamp
df_yesterday = (
    spark.read.format("delta")
    .option("timestampAsOf", "2026-03-09T00:00:00Z")
    .load("/tables/orders")
)

# SQL syntax
spark.sql("""
    SELECT * FROM catalog.silver.orders VERSION AS OF 5
""")

spark.sql("""
    SELECT * FROM catalog.silver.orders TIMESTAMP AS OF '2026-03-09'
""")

2.2 Inspecting Table History

python

from delta.tables import DeltaTable

dt = DeltaTable.forName(spark, "catalog.silver.orders")
history = dt.history()
history.select("version", "timestamp", "operation", "operationParameters", "operationMetrics").show(20, truncate=False)

# SQL alternative
spark.sql("DESCRIBE HISTORY catalog.silver.orders").show(20, truncate=False)

2.3 Restoring a Table

python

# Restore to a previous version (creates a new commit)
dt.restoreToVersion(5)

# Restore to a timestamp
dt.restoreToTimestamp("2026-03-09")

# SQL
spark.sql("RESTORE TABLE catalog.silver.orders TO VERSION AS OF 5")

3. Schema Enforcement and Evolution

3.1 Schema Enforcement (Default Behavior)

Delta rejects writes whose schema doesn't match the table:

python

# Table has columns: order_id, customer_id, quantity, order_ts

# This FAILS - 'amount' column doesn't exist in the target
df_with_new_col = df.withColumn("amount", F.lit(99.99))
df_with_new_col.write.format("delta").mode("append").saveAsTable("catalog.silver.orders")
# AnalysisException: A schema mismatch detected when writing to the Delta table.

3.2 Schema Evolution (Opt-In)

python

# Enable auto-merge for new columns
(
    df_with_new_col
    .write
    .format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .saveAsTable("catalog.silver.orders")
)

# Or set at the session level
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

3.3 Schema Evolution Rules

Change Type	Supported	Method
Add new column	Yes	`mergeSchema = true`
Widen type (int -> long)	Yes	`mergeSchema = true`
Rename column	No (use ALTER TABLE)	`ALTER TABLE ... RENAME COLUMN`
Drop column	No (use ALTER TABLE)	`ALTER TABLE ... DROP COLUMN` (DBR 11+)
Change type (string -> int)	No	Requires overwrite with `overwriteSchema = true`

python

# Overwrite schema entirely (destructive - use with caution)
(
    df_new_schema
    .write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("catalog.silver.orders")
)

# Column management via ALTER TABLE
spark.sql("ALTER TABLE catalog.silver.orders ADD COLUMNS (amount DOUBLE COMMENT 'Order total in USD')")
spark.sql("ALTER TABLE catalog.silver.orders RENAME COLUMN quantity TO qty")
spark.sql("ALTER TABLE catalog.silver.orders DROP COLUMN legacy_field")

4. MERGE Operations

MERGE is the cornerstone of incremental data pipelines. It supports UPSERT, delete, and complex conditional logic in a single atomic operation.

4.1 Basic UPSERT

python

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "catalog.silver.customers")
source = spark.table("catalog.bronze.customers_incoming")

(
    target.alias("t")
    .merge(
        source.alias("s"),
        "t.customer_id = s.customer_id"
    )
    .whenMatchedUpdate(set={
        "customer_name": "s.customer_name",
        "email": "s.email",
        "updated_at": "current_timestamp()",
    })
    .whenNotMatchedInsert(values={
        "customer_id": "s.customer_id",
        "customer_name": "s.customer_name",
        "email": "s.email",
        "created_at": "current_timestamp()",
        "updated_at": "current_timestamp()",
    })
    .execute()
)

4.2 UPSERT with Conditional Logic

python

(
    target.alias("t")
    .merge(
        source.alias("s"),
        "t.customer_id = s.customer_id"
    )
    .whenMatchedUpdate(
        condition="s.updated_at > t.updated_at",  # Only update if source is newer
        set={
            "customer_name": "s.customer_name",
            "email": "s.email",
            "updated_at": "s.updated_at",
        }
    )
    .whenMatchedDelete(
        condition="s.is_deleted = true"  # Soft delete processing
    )
    .whenNotMatchedInsert(
        condition="s.is_deleted = false",  # Don't insert already-deleted records
        values={
            "customer_id": "s.customer_id",
            "customer_name": "s.customer_name",
            "email": "s.email",
            "created_at": "current_timestamp()",
            "updated_at": "s.updated_at",
        }
    )
    .execute()
)

4.3 SCD Type 2 Implementation

Slowly Changing Dimension Type 2 preserves the full history of changes.

python

# Target schema:
# customer_id, customer_name, email, effective_from, effective_to, is_current

target = DeltaTable.forName(spark, "catalog.gold.dim_customer_scd2")
source = spark.table("catalog.silver.customers_changes")

# Step 1: Identify changed records
staged_updates = (
    source.alias("s")
    .join(
        target.toDF().alias("t"),
        (F.col("s.customer_id") == F.col("t.customer_id")) & (F.col("t.is_current") == True)
    )
    .where(
        (F.col("s.customer_name") != F.col("t.customer_name"))
        | (F.col("s.email") != F.col("t.email"))
    )
    .select(
        F.col("s.customer_id"),
        F.col("s.customer_name"),
        F.col("s.email"),
        F.col("s.updated_at").alias("effective_from"),
    )
)

# Step 2: Union new inserts with records to close
new_records = staged_updates.withColumn("effective_to", F.lit(None).cast("timestamp")).withColumn("is_current", F.lit(True))
records_to_close = staged_updates.select("customer_id", "effective_from")

# Step 3: Merge - close old records and insert new ones
(
    target.alias("t")
    .merge(
        new_records.alias("s"),
        "t.customer_id = s.customer_id AND t.is_current = true"
    )
    .whenMatchedUpdate(set={
        "effective_to": "s.effective_from",
        "is_current": "false",
    })
    .whenNotMatchedInsert(values={
        "customer_id": "s.customer_id",
        "customer_name": "s.customer_name",
        "email": "s.email",
        "effective_from": "s.effective_from",
        "effective_to": "s.effective_to",
        "is_current": "s.is_current",
    })
    .execute()
)

4.4 SQL MERGE Syntax

For simpler merge logic, SQL can be more readable:

sql

MERGE INTO catalog.silver.orders AS t
USING catalog.bronze.orders_incoming AS s
ON t.order_id = s.order_id

WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET
    t.quantity = s.quantity,
    t.amount = s.amount,
    t.updated_at = s.updated_at

WHEN NOT MATCHED THEN
  INSERT (order_id, customer_id, quantity, amount, created_at, updated_at)
  VALUES (s.order_id, s.customer_id, s.quantity, s.amount, current_timestamp(), s.updated_at)

WHEN NOT MATCHED BY SOURCE AND t.order_date < date_sub(current_date(), 365) THEN
  DELETE

5. Table Maintenance

5.1 VACUUM

VACUUM removes data files no longer referenced by the transaction log. Without it, storage costs grow unboundedly.

python

# Remove files older than 7 days (default retention)
dt = DeltaTable.forName(spark, "catalog.silver.orders")
dt.vacuum(retentionHours=168)  # 7 * 24

# SQL
spark.sql("VACUUM catalog.silver.orders RETAIN 168 HOURS")

# DRY RUN first to see what would be deleted
spark.sql("VACUUM catalog.silver.orders RETAIN 168 HOURS DRY RUN")

Critical warning: After VACUUM, time travel to versions older than the retention period will fail because the underlying files are gone.

python

# NEVER set retention below 7 days without understanding the consequences
# This safety check is enabled by default:
# spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")  # DANGEROUS

5.2 OPTIMIZE

OPTIMIZE compacts small files into larger ones, dramatically improving read performance.

python

# Optimize the entire table
spark.sql("OPTIMIZE catalog.silver.orders")

# Optimize a specific partition (recommended for large tables)
spark.sql("OPTIMIZE catalog.silver.orders WHERE order_date >= '2026-03-01'")

# Python API
dt.optimize().executeCompaction()

# With partition filter
dt.optimize().where("order_date >= '2026-03-01'").executeCompaction()

5.3 Z-ORDER

Z-ORDER co-locates related data within files for faster point lookups and range scans.

python

# Z-ORDER by frequently filtered columns
spark.sql("OPTIMIZE catalog.silver.orders ZORDER BY (customer_id, product_id)")

# Python API
dt.optimize().where("order_date >= '2026-03-01'").executeZOrderBy("customer_id", "product_id")

When to use Z-ORDER:

Use Z-ORDER When	Avoid Z-ORDER When
Columns in frequent WHERE clauses	Column is already a partition column
Columns used in JOIN keys	Table is small (< 1 GB)
High-cardinality columns	Column is rarely filtered on
Up to 3-4 columns	More than 4 columns (diminishing returns)

5.4 Liquid Clustering (Databricks, Delta 3.0+)

Liquid clustering is the next-generation replacement for both partitioning and Z-ORDER. It automatically reorganizes data as it evolves.

python

# Create a table with liquid clustering
spark.sql("""
    CREATE TABLE catalog.silver.orders_lc (
        order_id STRING,
        customer_id STRING,
        product_id STRING,
        quantity INT,
        amount DOUBLE,
        order_ts TIMESTAMP
    )
    USING DELTA
    CLUSTER BY (customer_id, order_ts)
""")

# Change clustering keys without rewriting (unlike partition changes!)
spark.sql("ALTER TABLE catalog.silver.orders_lc CLUSTER BY (order_ts, product_id)")

# Trigger clustering
spark.sql("OPTIMIZE catalog.silver.orders_lc")

Liquid clustering vs traditional approaches:

Feature	Partition + Z-ORDER	Liquid Clustering
Key changes	Requires full rewrite	ALTER TABLE
Incremental	Manual OPTIMIZE per partition	Automatic
Write performance	Can cause small files	Optimized writes
Filter columns	Limited by partition choice	Flexible
Maturity	GA everywhere	DBR 13.3+

5.5 Maintenance Schedule

python

# Recommended production maintenance job
def run_table_maintenance(table_name: str, partition_filter: str = None):
    """Run standard Delta maintenance operations."""
    optimize_sql = f"OPTIMIZE {table_name}"
    if partition_filter:
        optimize_sql += f" WHERE {partition_filter}"

    # Step 1: Compact files
    spark.sql(optimize_sql)

    # Step 2: Vacuum old files (7-day retention)
    spark.sql(f"VACUUM {table_name} RETAIN 168 HOURS")

    # Step 3: Analyze table statistics for optimizer
    spark.sql(f"ANALYZE TABLE {table_name} COMPUTE STATISTICS FOR ALL COLUMNS")

    print(f"Maintenance complete for {table_name}")

# Run for recent partitions only (efficient for large tables)
run_table_maintenance(
    "catalog.silver.orders",
    partition_filter="order_date >= current_date() - INTERVAL 7 DAYS"
)

6. Change Data Feed (CDF)

Change Data Feed captures row-level changes (inserts, updates, deletes) in a Delta table and exposes them as a queryable stream.

6.1 Enabling CDF

python

# Enable on existing table
spark.sql("ALTER TABLE catalog.silver.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true)")

# Enable at creation time
spark.sql("""
    CREATE TABLE catalog.silver.customers_cdf (
        customer_id STRING,
        customer_name STRING,
        email STRING
    )
    USING DELTA
    TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

6.2 Reading Changes

python

# Batch: read changes between versions
changes = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 5)
    .option("endingVersion", 10)
    .table("catalog.silver.customers")
)

changes.select(
    "_change_type",  # insert, update_preimage, update_postimage, delete
    "_commit_version",
    "_commit_timestamp",
    "customer_id",
    "customer_name",
).show(truncate=False)

# Streaming: continuous change consumption
changes_stream = (
    spark.readStream.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", "latest")
    .table("catalog.silver.customers")
)

6.3 CDF Use Cases

python

# Use case: Propagate changes to gold aggregation table
def process_customer_changes(batch_df, batch_id):
    """Process CDF micro-batch to update gold customer metrics."""
    # Filter to only the latest state (post-image for updates, inserts, deletes)
    relevant = batch_df.where(
        F.col("_change_type").isin("insert", "update_postimage", "delete")
    )

    # Merge into gold table
    gold_table = DeltaTable.forName(spark, "catalog.gold.customer_metrics")
    (
        gold_table.alias("g")
        .merge(
            relevant.alias("s"),
            "g.customer_id = s.customer_id"
        )
        .whenMatchedDelete(condition="s._change_type = 'delete'")
        .whenMatchedUpdate(set={
            "customer_name": "s.customer_name",
            "last_updated": "s._commit_timestamp",
        })
        .whenNotMatchedInsert(values={
            "customer_id": "s.customer_id",
            "customer_name": "s.customer_name",
            "last_updated": "s._commit_timestamp",
        })
        .execute()
    )

# Run as streaming pipeline
(
    changes_stream
    .writeStream
    .foreachBatch(process_customer_changes)
    .option("checkpointLocation", "/checkpoints/customer_cdf_to_gold")
    .trigger(availableNow=True)
    .start()
)

Hands-On Exercises

Exercise 1: Time Travel Audit Trail

1. Create a Delta table workshop.sandbox.audit_test with 100 rows

2. Run 5 successive updates, each modifying a subset of rows

3. Query the table history and note the versions

4. Use time travel to compare row counts at version 0 vs version 5

5. Restore the table to version 2 and verify the state

python

# Starter code
from pyspark.sql import functions as F

# Create initial data
initial = spark.range(100).withColumn("value", (F.rand() * 1000).cast("int"))
initial.write.format("delta").mode("overwrite").saveAsTable("workshop.sandbox.audit_test")

# YOUR CODE: Run 5 updates
# YOUR CODE: Query history
# YOUR CODE: Time travel comparison
# YOUR CODE: Restore to version 2

Exercise 2: Production MERGE Pipeline

Build a complete UPSERT pipeline:

1. Create a target table workshop.sandbox.products with 1000 products

2. Generate an "incoming" DataFrame with 200 products: 150 existing (with changes) and 50 new

3. Implement a MERGE that updates changed records and inserts new ones

4. Add a last_synced_at timestamp to track when each record was last processed

5. Verify the merge metrics from the operation result

python

# Starter code
product_schema = "product_id STRING, name STRING, category STRING, price DOUBLE"

# Step 1: Create target with 1000 rows
# Step 2: Generate incoming batch (150 updates + 50 inserts)
# Step 3: MERGE with conditional update
# Step 4: Verify metrics

Exercise 3: Maintenance Automation

1. Create a table with intentionally poor file layout (1000 small files)

2. Run DESCRIBE DETAIL to check file count and size

3. Run OPTIMIZE and observe the improvement

4. Apply Z-ORDER on two columns

5. Run VACUUM with a dry run, then the actual vacuum

6. Compare query performance before and after maintenance

Exercise 4: Change Data Feed Pipeline

1. Create a CDF-enabled table workshop.sandbox.inventory

2. Insert initial stock levels for 50 products

3. Run 3 rounds of updates (stock adjustments)

4. Read the CDF to build an audit log of all inventory movements

5. Compute net stock change per product from the CDF data

Key Takeaways

1. Delta Lake's transaction log is the foundation for ACID guarantees. Every write is a commit that is atomic and durable.

2. Time travel is not just for debugging - use it for auditing, reproducible ML training sets, and safe rollbacks.

3. Schema enforcement protects your tables by default. Opt into schema evolution explicitly with mergeSchema when you need it.

4. MERGE is the Swiss Army knife of incremental processing. Master basic UPSERT, conditional logic, and SCD Type 2 patterns.

5. Table maintenance is mandatory in production. Schedule OPTIMIZE, VACUUM, and ANALYZE TABLE regularly.

6. Z-ORDER on 2-4 high-cardinality filter columns can give 10x+ query speedups. Liquid clustering is the future.

7. Change Data Feed enables efficient CDC pipelines. Use it to propagate changes between layers without full table scans.

Knowledge Check

1. What happens if you run VACUUM with a 0-hour retention? Why is the safety check important?

2. A table has 10,000 small files (1 MB each). What is the impact on query performance, and which operations fix it?

3. Explain the difference between mergeSchema and overwriteSchema. When would you use each?

4. In an SCD Type 2 MERGE, why do you need to both UPDATE existing records and INSERT new ones?

5. Your MERGE operation is running for 2 hours on a 100M-row target with 500K-row source. What are three things you would check to improve performance?

6. What are the _change_type values in a CDF read, and what does update_preimage represent?

Chapter 3

🔒 Available in full product

Chapter 03: Medallion Architecture in Practice

Chapter 4

🔒 Available in full product

Chapter 04: Data Pipeline Orchestration

Chapter 5

🔒 Available in full product

Chapter 05: Unity Catalog & Data Governance

Chapter 6

🔒 Available in full product

Chapter 06: Testing & Data Quality

Chapter 7

🔒 Available in full product

Chapter 07: CI/CD for Data Pipelines

Chapter 8

🔒 Available in full product

Chapter 08: Production Operations & Performance

You’ve reached the end of the free preview

Get the full Databricks Data Engineering Mastery and unlock everything.

All Chapters

Get the complete guide with every chapter unlocked, including code samples, diagrams, and best practices.

Full Tool Suite

Access all interactive tools with complete data, all workload profiles, and the full scenario library.

Source Files

Downloadable source code, configuration files, and working examples from every chapter.

Lifetime Updates

Free updates for life. Every new chapter, tool, and improvement included.

Buy Now — $149 →
📦 Free sample included — download another copy for the full product.

Databricks Data Engineering Mastery v1.0.0 — Free Preview

Contents

Chapter 01: PySpark Foundations for Data Engineers

Learning Objectives

1. SparkSession: Your Entry Point to Spark

1.1 Creating a SparkSession

1.2 Key Configuration Properties

1.3 SparkSession on Databricks

2. The DataFrame API Deep-Dive

2.1 Creating DataFrames

2.2 Core Transformations

2.3 Common DataFrame Operations

2.4 Joins

3. Column Expressions

3.1 The F.col() Pattern

3.2 Conditional Logic with when/otherwise

3.3 Working with Complex Types

3.4 Date and Timestamp Functions

4. UDFs vs Built-in Functions

4.1 Why UDFs Are Expensive

4.2 When UDFs Are Acceptable

4.3 Pandas UDFs: The Middle Ground

4.4 Decision Framework

5. Lazy Evaluation and Execution Plans

5.1 Lazy Evaluation

5.2 Reading Execution Plans

5.3 The Catalyst Optimizer

6. Partitioning Basics

6.1 In-Memory Partitions vs On-Disk Partitions

6.2 Controlling In-Memory Partitions

6.3 On-Disk Partitioning

6.4 The Small File Problem

Hands-On Exercises

Exercise 1: DataFrame Gymnastics

Exercise 2: UDF Elimination

Exercise 3: Execution Plan Analysis

Exercise 4: Partition Tuning

Key Takeaways

Knowledge Check

Chapter 02: Delta Lake Deep Dive

Learning Objectives

1. Delta Lake Architecture

1.1 What Makes Delta Different from Parquet

1.2 The Transaction Log Protocol

1.3 ACID Transactions

2. Time Travel

2.1 Querying by Version and Timestamp

2.2 Inspecting Table History

2.3 Restoring a Table

3. Schema Enforcement and Evolution

3.1 Schema Enforcement (Default Behavior)

3.2 Schema Evolution (Opt-In)

3.3 Schema Evolution Rules

4. MERGE Operations

4.1 Basic UPSERT

4.2 UPSERT with Conditional Logic

4.3 SCD Type 2 Implementation

4.4 SQL MERGE Syntax

5. Table Maintenance

5.1 VACUUM

5.2 OPTIMIZE

5.3 Z-ORDER

5.4 Liquid Clustering (Databricks, Delta 3.0+)

5.5 Maintenance Schedule

6. Change Data Feed (CDF)

6.1 Enabling CDF

6.2 Reading Changes

6.3 CDF Use Cases

Hands-On Exercises

Exercise 1: Time Travel Audit Trail

Exercise 2: Production MERGE Pipeline

Exercise 3: Maintenance Automation

Exercise 4: Change Data Feed Pipeline

Key Takeaways

Knowledge Check

Chapter 03: Medallion Architecture in Practice

Chapter 04: Data Pipeline Orchestration

Chapter 05: Unity Catalog & Data Governance

Chapter 06: Testing & Data Quality

Chapter 07: CI/CD for Data Pipelines

Chapter 08: Production Operations & Performance

3.1 The `F.col()` Pattern

3.2 Conditional Logic with `when`/`otherwise`