DEV Community: Karan Pulyala

Delta Lake Change Data Feed on Azure Databricks: readChangeFeed, Incremental MERGE, and the Retention Traps

Karan Pulyala — Thu, 29 Oct 2026 14:17:54 +0000

Most Silver-to-Gold jobs I've seen start the same way: re-read the whole Silver table, recompute everything, overwrite Gold. It works until Silver hits a few billion rows and the "daily" job takes four hours to process the 0.1% of rows that actually changed.

Delta Lake's Change Data Feed (CDF) fixes that. It lets you read only the rows that were inserted, updated, or deleted between two table versions, with the type of change attached. This post covers how to turn it on, how to read it in batch and streaming, how to turn a change feed into a correct MERGE, and the retention and schema traps that show up in interviews and in production.

Turning it on

CDF is a table property. You can set it when you create the table or add it later:

-- New table
CREATE TABLE silver.orders (
  order_id    BIGINT,
  customer_id BIGINT,
  status      STRING,
  amount      DECIMAL(18,2),
  updated_at  TIMESTAMP
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- Existing table
ALTER TABLE silver.orders
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

Two things to know straight away. First, changes are only recorded from the version where you enabled it. You can't go back and get a change feed for history from before that. Second, you can make it the default for new tables in a session with spark.databricks.delta.properties.defaults.enableChangeDataFeed = true, which is handy in a pipeline that creates many tables.

What a change row looks like

When you read the feed, you get the table's columns plus three metadata columns:

_change_type: one of insert, update_preimage, update_postimage, or delete
_commit_version: the Delta version that made the change
_commit_timestamp: when that commit happened

An UPDATE or a matched MERGE produces two rows per changed record: the old values (update_preimage) and the new values (update_postimage). That pair is the most useful part of CDF and also the most common source of bugs, which we'll get to.

Batch reads

changes = (
    spark.read
    .option("readChangeFeed", "true")
    .option("startingVersion", 12)
    .option("endingVersion", 15)      # optional, inclusive
    .table("silver.orders")
)

changes.groupBy("_change_type").count().show()

You can use startingTimestamp and endingTimestamp instead of versions. In SQL, the same thing is a table-valued function:

SELECT _change_type, COUNT(*)
FROM table_changes('silver.orders', 12, 15)
GROUP BY _change_type;

Both bounds are inclusive. If you track "last processed version" yourself, start the next run at last_version + 1, not last_version, or you'll process the boundary commit twice.

To find the current version for your bookkeeping:

latest = spark.sql("DESCRIBE HISTORY silver.orders LIMIT 1").first()["version"]

Streaming reads (usually the better choice)

Hand-rolled version tracking works, but a streaming read lets the checkpoint do the bookkeeping for you:

stream = (
    spark.readStream
    .option("readChangeFeed", "true")
    .option("startingVersion", 12)   # only used on the first run
    .table("silver.orders")
)

startingVersion only matters the first time the query starts with an empty checkpoint. After that, the checkpoint decides where to resume. Combined with trigger(availableNow=True), you get an incremental batch job that processes exactly what changed since the last run and then stops.

Turning changes into a correct MERGE

Here's the trap. Between two runs, the same order_id might have been updated three times, or inserted and then deleted. If you MERGE the raw feed into Gold, you get this error:

Cannot perform Merge as multiple source rows matched and attempted to modify the same target row

Or worse, if you deduplicate carelessly, you apply an old update after a newer one. The fix is to drop the pre-images and keep only the latest change per key:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

def latest_changes(df):
    w = Window.partitionBy("order_id").orderBy(F.col("_commit_version").desc())
    return (
        df.filter(F.col("_change_type") != "update_preimage")
          .withColumn("rn", F.row_number().over(w))
          .filter("rn = 1")
          .drop("rn")
    )

Within a single commit, a key normally has at most one post-image or delete, so _commit_version is enough to order them. Then upsert with foreachBatch:

from delta.tables import DeltaTable

def upsert_to_gold(batch_df, batch_id):
    src = latest_changes(batch_df)
    gold = DeltaTable.forName(spark, "gold.orders_current")
    (
        gold.alias("t")
        .merge(src.alias("s"), "t.order_id = s.order_id")
        .whenMatchedDelete(condition="s._change_type = 'delete'")
        .whenMatchedUpdate(
            condition="s._change_type != 'delete'",
            set={
                "customer_id": "s.customer_id",
                "status": "s.status",
                "amount": "s.amount",
                "updated_at": "s.updated_at",
            },
        )
        .whenNotMatchedInsert(
            condition="s._change_type != 'delete'",
            values={
                "order_id": "s.order_id",
                "customer_id": "s.customer_id",
                "status": "s.status",
                "amount": "s.amount",
                "updated_at": "s.updated_at",
            },
        )
        .execute()
    )

(
    stream.writeStream
    .foreachBatch(upsert_to_gold)
    .option("checkpointLocation", "/Volumes/main/gold/_checkpoints/orders_current")
    .trigger(availableNow=True)
    .start()
)

Notice the condition on the insert: if a key was inserted and deleted inside the same batch, the latest change is a delete, and the row should never reach Gold.

Where pre-images earn their keep: aggregates

For a plain mirror, you throw pre-images away. For aggregates, they're exactly what you need. Say Gold keeps revenue per customer. An update from 100 to 120 shouldn't add 120, it should add 20. Pre-images give you the subtraction:

signed = (
    batch_df
    .withColumn(
        "delta_amount",
        F.when(F.col("_change_type").isin("insert", "update_postimage"), F.col("amount"))
         .when(F.col("_change_type").isin("delete", "update_preimage"), -F.col("amount"))
    )
    .groupBy("customer_id")
    .agg(F.sum("delta_amount").alias("delta_amount"))
)

Then MERGE that into the aggregate table with t.revenue = t.revenue + s.delta_amount. You update a handful of customers instead of re-aggregating the whole table. Be careful about idempotency here: an additive merge applied twice double-counts. foreachBatch can replay a batch after a failure, so either make the write idempotent with Delta's txnAppId/txnVersion options (on appends) or keep the aggregate recomputable from Silver as a fallback.

The traps interviewers ask about

Retention. Change data lives in a _change_data folder inside the table (for some operations, like insert-only appends, Delta derives the changes from the regular data files instead). Those files are removed by VACUUM like any other, under delta.deletedFileRetentionDuration, which defaults to 7 days. If your consumer is down for two weeks, the versions it needs may be gone, and the read fails. Size the retention to your worst realistic outage, not your normal schedule.

Reading before CDF was enabled. Asking for a version from before you turned it on is an error, not an empty result. Start from the version where the property was set (you'll find it in DESCRIBE HISTORY as a SET TBLPROPERTIES operation).

Schema changes. Additive changes such as a new column are fine. Non-additive changes on tables with column mapping, like renaming or dropping a column, can block batch CDF reads over a range that spans the change. Plan those changes, and restart the downstream stream deliberately afterwards.

Overwrites. A full overwrite of the source shows up as deletes of every old row and inserts of every new row. Your incremental job then processes the entire table. If an upstream job overwrites "to be safe", CDF won't save you; switching it to MERGE or replaceWhere will.

Cost. Enabling CDF adds a little write overhead for updates, deletes, and merges, because Delta writes the change files too. For tables nobody reads incrementally, leave it off.

Interview questions to practise

What are the four _change_type values, and which operations produce update_preimage?
Why can't you MERGE a raw change feed straight into a target table?
A CDF consumer was paused for 10 days and now fails on start. What probably happened?
When would you keep pre-images instead of dropping them?
How is reading CDF as a stream different from tracking startingVersion yourself?

If you can explain question 2 with a concrete example of three updates to one key, you've got the core idea.

If you want more practice like this, there's a free 10-question sample of Azure Databricks and PySpark interview questions with worked answers: free sample. The full set is in the Azure Databricks & PySpark Interview Pack.

Spark JDBC Parallel Reads on Azure Databricks: partitionColumn, Bounds, Predicates, and Not Knocking Over Your Database

Karan Pulyala — Wed, 28 Oct 2026 14:22:40 +0000

Pulling a big table out of Azure SQL Database (or any relational source) with Spark looks like a one-liner. Then the job runs for two hours on a 16-node cluster and the Spark UI shows exactly one task doing all the work. This post walks through why that happens and how to read and write over JDBC in parallel without hurting the source database. It's also a favourite interview topic, because it tests whether you understand how Spark actually splits work.

Why the default JDBC read uses one task

jdbc_url = (
    "jdbc:sqlserver://myserver.database.windows.net:1433;"
    "database=sales;encrypt=true;trustServerCertificate=false;"
)
props = {
    "user": dbutils.secrets.get("kv-scope", "sql-user"),
    "password": dbutils.secrets.get("kv-scope", "sql-password"),
}

df = (spark.read.format("jdbc")
      .option("url", jdbc_url)
      .option("dbtable", "dbo.orders")
      .options(**props)
      .load())

print(df.rdd.getNumPartitions())  # 1

Spark has no idea how to split an arbitrary SQL table, so without extra options it opens one connection and runs one SELECT over the whole table. One partition means one task on one core, no matter how big the cluster is. Everything downstream inherits that single partition until a shuffle happens.

(Keep credentials in a Databricks secret scope backed by Key Vault, as above. Never hard-code them in a notebook.)

Parallel reads with partitionColumn

To get parallelism, you give Spark four options together:

bounds = (spark.read.format("jdbc")
          .option("url", jdbc_url)
          .option("query", "SELECT MIN(order_id) AS lo, MAX(order_id) AS hi FROM dbo.orders")
          .options(**props)
          .load()
          .first())

orders = (spark.read.format("jdbc")
          .option("url", jdbc_url)
          .option("dbtable", "dbo.orders")
          .option("partitionColumn", "order_id")
          .option("lowerBound", bounds["lo"])
          .option("upperBound", bounds["hi"] + 1)
          .option("numPartitions", 16)
          .option("fetchsize", 10000)
          .options(**props)
          .load())

print(orders.rdd.getNumPartitions())  # 16

Spark divides the range between lowerBound and upperBound into numPartitions equal strides and runs one query per stride, each over its own connection. Roughly:

-- partition 0
SELECT ... FROM dbo.orders WHERE order_id < 625001 OR order_id IS NULL
-- partitions 1..14
SELECT ... FROM dbo.orders WHERE order_id >= 625001 AND order_id < 1250001
-- partition 15
SELECT ... FROM dbo.orders WHERE order_id >= 9375001

A few rules that trip people up:

The bounds don't filter anything. The first partition gets everything below the first stride (plus NULLs) and the last gets everything above the final one. If your bounds are wrong you don't lose rows; you get one enormous partition instead. That's why querying MIN/MAX first is worth the extra round trip.
partitionColumn must be numeric, date, or timestamp. Strings don't work.
You can't combine query with partitionColumn. If you need a filtered or joined source, wrap it as a subquery in dbtable: .option("dbtable", "(SELECT ... FROM dbo.orders WHERE status = 'SHIPPED') AS src").
Pick an indexed column. Each partition is a range predicate. On a clustered index key it's a cheap seek; on an unindexed column, 16 partitions means 16 full scans of the table.

The skew problem: equal ranges, unequal rows

Strides are equal in value, not in row count. If order_id has a big gap (say IDs jumped from 2 million to 50 million after a migration), most strides return nothing and a couple return everything. You'll see it in the Spark UI as a handful of long tasks while the rest finish in seconds.

When the key isn't evenly distributed, use explicit predicates instead. Each predicate becomes exactly one partition, so you control the split:

predicates = [
    f"order_date >= '{m}-01' AND order_date < DATEADD(month, 1, '{m}-01')"
    for m in ["2026-01", "2026-02", "2026-03", "2026-04", "2026-05", "2026-06"]
]

orders_h1 = spark.read.jdbc(
    url=jdbc_url,
    table="dbo.orders",
    predicates=predicates,
    properties=props,
)

print(orders_h1.rdd.getNumPartitions())  # 6

Two things to check with predicates: they must not overlap (or you duplicate rows), and together they must cover everything you want (or you silently drop rows). Unlike the stride approach, there's no automatic catch-all for NULLs or out-of-range values. A quick count() against a source-side COUNT(*) is a cheap sanity test.

Don't knock over the source

numPartitions is also the number of concurrent connections Spark opens against the database. Sixty-four partitions against a small Azure SQL tier will throttle on DTU/vCore limits or hit the session cap, and the production app sharing that database will feel it. In practice:

Start at 8 to 16 partitions and watch the database's CPU and log IO metrics, not just Spark's runtime.
Read from a readable secondary or a replica when one exists (for Azure SQL, add applicationIntent=ReadOnly to the URL on tiers that support read scale-out).
Schedule big extracts outside business hours.

fetchsize, pushdown, and reading only what you need

fetchsize controls how many rows the driver pulls per round trip. Small values mean thousands of network round trips per partition; some drivers (Oracle's defaults to 10) are painfully conservative. Values in the 5,000 to 20,000 range are a reasonable start for typical row widths.

Filters you apply after load() are pushed into the SQL by default (pushDownPredicate is true), and only the selected columns are fetched. You can confirm it with explain():

(orders
 .filter("status = 'SHIPPED'")
 .select("order_id", "customer_id", "amount")
 .explain())
# Look for PushedFilters: [*IsNotNull(status), *EqualTo(status,SHIPPED)]
# and ReadSchema with only the three columns

If a filter doesn't appear under PushedFilters (for example, one built on a Python UDF), Spark pulls every row and filters it afterwards. For incremental loads, filter on a watermark column such as modified_at > last_run so each run moves only changed rows, then land them in Delta with MERGE.

Writing back over JDBC

Writes are parallel by default: each DataFrame partition writes over its own connection. That's the risk here, because 400 partitions after a shuffle means 400 connections hammering the target.

(summary_df
 .write.format("jdbc")
 .option("url", jdbc_url)
 .option("dbtable", "dbo.daily_summary")
 .option("numPartitions", 8)     # Spark coalesces down to 8 before writing
 .option("batchsize", 10000)     # rows per INSERT batch, default 1000
 .option("truncate", "true")     # with overwrite: TRUNCATE instead of DROP/CREATE
 .options(**props)
 .mode("overwrite")
 .save())

truncate matters more than people expect. Without it, overwrite drops and recreates the table, losing indexes, permissions, and column types you defined in the database. Also remember JDBC writes aren't transactional across partitions: if task 5 of 8 fails, partitions 1 to 4 are already committed. For anything that must be all-or-nothing, write to a staging table and swap or MERGE on the database side in one statement.

When not to use plain JDBC

On Databricks, Lakehouse Federation lets you query SQL Server and other sources through Unity Catalog without managing JDBC options per notebook, and Lakeflow Connect covers managed ingestion for some sources. For large one-off extracts, exporting to files in ADLS and reading those can be faster than any number of JDBC partitions. JDBC is still the right tool for targeted incremental pulls and small write-backs, and it's what most interview questions are about.

Interview questions to practise

Why does a default JDBC read produce one partition, and what four options fix it?
Do lowerBound and upperBound filter rows? What happens to IDs outside them?
Your parallel JDBC read has two tasks taking 90% of the time. What's going on and how do you fix it?
How do you stop a Spark write from opening hundreds of connections to Azure SQL?
What does truncate=true change about mode("overwrite") over JDBC?

If you can explain the generated WHERE clauses for question 2 from memory, you understand this better than most candidates.

Delta Lake Schema Evolution and Time Travel on Azure Databricks: mergeSchema, overwriteSchema, Column Mapping, and RESTORE

Karan Pulyala — Tue, 27 Oct 2026 14:25:39 +0000

Sooner or later a source system adds a column, renames one, or someone runs a bad overwrite on a production table. How a Delta table reacts in those moments is a favourite interview topic, because it separates people who have used Delta Lake from people who have only read about it. This post covers schema enforcement, the two schema evolution options, column mapping for renames and drops, and how to use time travel and RESTORE to get out of trouble on Azure Databricks.

Schema enforcement: Delta says no by default

Delta checks every write against the table's schema. If you append a DataFrame with a column the table doesn't have, the write fails instead of silently dropping or adding data.

from pyspark.sql import functions as F

orders = spark.createDataFrame(
    [(1, "C001", 120.50), (2, "C002", 80.00)],
    "order_id INT, customer_id STRING, amount DOUBLE",
)
orders.write.format("delta").mode("overwrite").saveAsTable("demo.orders")

# The source now sends a new column
new_orders = spark.createDataFrame(
    [(3, "C003", 42.00, "web")],
    "order_id INT, customer_id STRING, amount DOUBLE, channel STRING",
)

new_orders.write.format("delta").mode("append").saveAsTable("demo.orders")
# AnalysisException: A schema mismatch detected when writing to the Delta table...

The same check catches type mismatches. Writing a string into an INT column fails rather than quietly casting it to null. That is the behaviour you want in Silver and Gold tables. A missing column is fine, though: if the incoming DataFrame lacks a nullable column, Delta writes null for it.

Option 1: mergeSchema adds columns

When the new column is expected, opt in to evolution for that one write:

(new_orders.write
    .format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .saveAsTable("demo.orders"))

spark.table("demo.orders").printSchema()
# order_id, customer_id, amount, channel  (old rows have channel = null)

mergeSchema adds new columns to the end of the schema and can make some compatible changes, such as letting a NOT NULL column accept nulls. It does not let you change amount from DOUBLE to STRING. Type changes need a rewrite (or, on recent runtimes, the type widening table feature for safe widenings like INT to BIGINT).

You can also turn evolution on for the whole session with spark.databricks.delta.schema.autoMerge.enabled = true. I'd avoid that in shared jobs. It applies to every write in the session, so an upstream bug that adds a junk column lands in your table without anyone noticing.

Schema evolution in MERGE

MERGE is where people get caught. A per-write mergeSchema option doesn't apply to it. On recent Databricks runtimes you can ask for evolution in the statement itself:

MERGE WITH SCHEMA EVOLUTION INTO demo.orders AS t
USING updates AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

On older runtimes you set the autoMerge session config just before the merge. Either way, evolution only happens with UPDATE SET * and INSERT * (or with explicit assignments to the new column); a merge that lists only existing columns won't add anything.

Option 2: overwriteSchema replaces everything

Sometimes you really want a new schema, for example after a redesign of a Gold table:

redesigned = (spark.table("demo.orders")
    .withColumn("amount", F.col("amount").cast("decimal(12,2)"))
    .drop("channel"))

(redesigned.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("demo.orders"))

This rewrites the data and replaces the schema, including types and partitioning. It's the right tool for intentional redesigns and the wrong one for daily jobs. Any downstream query that depended on channel breaks on its next run. In SQL, CREATE OR REPLACE TABLE ... AS SELECT does the same job. Both keep the table's history, so you can still time travel to before the change, which is why they beat DROP TABLE followed by a fresh CREATE.

Renaming and dropping columns: column mapping

Delta originally matched columns to Parquet files by physical name, so renaming a column meant rewriting the table. Column mapping fixes that by giving each column a stable internal ID:

ALTER TABLE demo.orders SET TBLPROPERTIES (
  'delta.columnMapping.mode' = 'name'
);

ALTER TABLE demo.orders RENAME COLUMN customer_id TO customer_key;
ALTER TABLE demo.orders DROP COLUMN channel;

Both are metadata-only operations, so they're instant even on huge tables. Two things to mention in an interview. First, enabling column mapping upgrades the table protocol, so old readers and some external tools that read the Parquet files directly may no longer read it correctly. Second, DROP COLUMN doesn't remove the data from the files until they are rewritten, so it's not how you remove sensitive data.

Time travel: reading the past

Every commit to a Delta table creates a version in the _delta_log. You can look at them:

DESCRIBE HISTORY demo.orders;

The output shows the version, timestamp, operation (WRITE, MERGE, DELETE...), the user, and operationMetrics such as rows and files written. Then query any version that hasn't been cleaned up:

SELECT * FROM demo.orders VERSION AS OF 3;
SELECT * FROM demo.orders TIMESTAMP AS OF '2026-10-26 08:00:00';

In PySpark:

v3 = (spark.read.format("delta")
      .option("versionAsOf", 3)
      .table("demo.orders"))

A handy pattern is diffing two versions to see what a job actually changed:

before = spark.read.option("versionAsOf", 5).table("demo.orders")
after  = spark.read.option("versionAsOf", 6).table("demo.orders")

removed = before.exceptAll(after)
added   = after.exceptAll(before)
print(removed.count(), added.count())

Time travel also makes reproducible reads easy. Pin a version for a model training run or a month-end report and you'll get the same data even after later loads.

RESTORE: undoing a bad write

Someone ran an overwrite with the wrong filter. Find the last good version in DESCRIBE HISTORY, then:

RESTORE TABLE demo.orders TO VERSION AS OF 5;
-- or
RESTORE TABLE demo.orders TO TIMESTAMP AS OF '2026-10-27 06:00:00';

RESTORE doesn't delete history. It writes a new commit that makes the table look like version 5, so you can even restore the restore if you picked the wrong version. It reports how many files were removed and restored. Restoring the table's data doesn't undo side effects elsewhere, though. If downstream jobs already read the bad version, they need to be rerun.

Why time travel isn't a backup

Time travel only works while both the log entries and the data files exist:

delta.logRetentionDuration (30 days by default) controls how long commit history is kept.
delta.deletedFileRetentionDuration (7 days by default) controls how old a removed file must be before VACUUM can delete it.

Once VACUUM runs, versions older than the file retention window usually can't be read anymore, because their files are gone. So by default you can count on about a week of time travel, not a month. If you need longer, raise both properties together and accept the storage cost:

ALTER TABLE demo.orders SET TBLPROPERTIES (
  'delta.logRetentionDuration' = 'interval 60 days',
  'delta.deletedFileRetentionDuration' = 'interval 30 days'
);

Time travel also lives in the same storage account as the table. If the container is deleted, history goes with it. For real recovery, use DEEP CLONE to a separate location or storage-level backups.

Interview questions to practise

What happens when you append a DataFrame with an extra column to a Delta table, and what are two ways to make it succeed?
Why is enabling autoMerge for a whole session risky?
How do you rename a column on a 5 TB Delta table without rewriting it, and what's the trade-off?
A table was overwritten by mistake yesterday. Walk through recovering it.
Why might VERSION AS OF fail for a version from 10 days ago?

If you can answer those with DESCRIBE HISTORY output in front of you, you're in good shape for most Delta Lake questions.

Repartition vs Coalesce vs partitionBy on Azure Databricks: Controlling Output Files When Writing Delta Tables

Karan Pulyala — Mon, 26 Oct 2026 14:23:04 +0000

Almost every PySpark job ends with a write, and that write is where a lot of quiet damage happens: ten thousand tiny files, one giant file written by a single task, or a job that suddenly takes an hour because someone added coalesce(1). Interviewers love this area because it tests whether you understand what a partition actually is. This post walks through repartition(), coalesce(), partitionBy(), and the settings that control how many files land in your Delta table on Azure Databricks.

Three different things called "partition"

Most confusion comes from one word meaning three things:

In-memory partitions: the chunks of a DataFrame that Spark processes, one task per partition.
Shuffle partitions: how many partitions Spark creates after a wide operation like groupBy or join, controlled by spark.sql.shuffle.partitions.
Table (directory) partitions: the folders created by df.write.partitionBy("col"), like event_date=2026-10-26/.

repartition() and coalesce() change the first. partitionBy() controls the third. The number of files you get is a product of both.

You can always check the first one:

from pyspark.sql import functions as F

df = spark.read.format("delta").load("/mnt/bronze/events")
print(df.rdd.getNumPartitions())

# Rows per in-memory partition, useful for spotting skew
(df.groupBy(F.spark_partition_id().alias("pid"))
   .count()
   .orderBy(F.desc("count"))
   .show(10))

When reading files, the number of input partitions is driven mostly by spark.sql.files.maxPartitionBytes (128 MB by default), not by how many files exist.

repartition(): a full shuffle, on purpose

repartition() always shuffles. It has three useful forms:

# 1. Round-robin into exactly 64 partitions (even sizes, no ordering)
df64 = df.repartition(64)

# 2. Hash partition by a column (rows with the same key end up together)
by_cust = df.repartition("customer_id")          # uses spark.sql.shuffle.partitions
by_cust_32 = df.repartition(32, "customer_id")   # explicit count

# 3. Range partition (sorted ranges, good before writing sorted output)
by_ts = df.repartitionByRange(32, "event_ts")

Use it when you need to increase parallelism (for example, after reading one large gzip file, which Spark can't split), when you need data co-located by key before a write, or when partitions are badly uneven and you want to rebalance.

The cost is a full shuffle: every row is serialized, sent over the network, and written to shuffle files. On a big table that's not free, so don't sprinkle repartition() around "just in case".

One subtle point: hash partitioning by a skewed key doesn't fix skew. If one customer_id has 40% of the rows, repartition("customer_id") puts all of them in one partition.

coalesce(): cheap, but it can shrink your whole stage

coalesce(n) only reduces the partition count, and it does it without a shuffle by merging existing partitions. That makes it a narrow transformation, which sounds like pure win. The catch is that Spark pipelines narrow transformations into the same stage, so the reduced parallelism can flow backwards into the work before it.

# Looks harmless, but the expensive transform now runs in only 1 task
(df.filter(F.col("country") == "CA")
   .withColumn("score", expensive_udf("payload"))
   .coalesce(1)
   .write.mode("overwrite").parquet("/mnt/tmp/out"))

Because there's no shuffle boundary between the read, the UDF, and the coalesce(1), the whole stage runs as one task. If you really need a single output file, repartition(1) is often faster here: the upstream work stays parallel and only the final write collapses to one task.

Rule of thumb:

Reducing a lot of partitions after heavy work, where the data has already shrunk (for example after a selective filter following a shuffle): coalesce() is fine.
Reducing to a very small number right after expensive narrow work: prefer repartition().
Needing more partitions: only repartition() can do that. coalesce(1000) on a 10-partition DataFrame still gives you 10.

Shuffle partitions and AQE

After a join or aggregation, Spark creates spark.sql.shuffle.partitions partitions, 200 by default in open-source Spark. For a 2 GB shuffle that's about 10 MB each; for a 2 TB shuffle it's 10 GB each, and tasks start spilling.

Adaptive Query Execution (on by default in Databricks) helps a lot by coalescing small shuffle partitions at runtime:

spark.conf.get("spark.sql.adaptive.enabled")                       # 'true'
spark.conf.get("spark.sql.adaptive.coalescePartitions.enabled")    # 'true'

# Target size AQE aims for when merging shuffle partitions
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

AQE can merge partitions, but it can't split one shuffle partition into more than it started with (apart from skew-join handling). So a common approach is to set the initial count high enough for your largest shuffles and let AQE merge them down. On Databricks you can also set spark.sql.shuffle.partitions to auto and let the platform pick the initial number.

partitionBy(): where file explosions come from

Here's the classic small-files trap:

(df.write
   .format("delta")
   .partitionBy("event_date")
   .mode("append")
   .saveAsTable("silver.events"))

Each task writes a separate file for every event_date value it contains. If the DataFrame has 200 partitions and the data spans 90 dates spread across all of them, you can get up to 200 × 90 = 18,000 files from one write.

The fix is to cluster rows by the partition column before writing, so each date is handled by one (or a few) tasks:

(df.repartition("event_date")
   .write
   .format("delta")
   .partitionBy("event_date")
   .mode("append")
   .saveAsTable("silver.events"))

Now each date goes to a single task and usually a single file. That creates the opposite risk: a big date becomes one huge file written by one slow task. Two ways to balance that:

# Cap rows per file so big partitions are split into several files
(df.repartition("event_date")
   .write
   .format("delta")
   .option("maxRecordsPerFile", 5_000_000)
   .partitionBy("event_date")
   .mode("append")
   .saveAsTable("silver.events"))

# Or spread each date over a few tasks with a salt
(df.repartition("event_date", (F.rand() * 4).cast("int"))
   .write.format("delta")
   .partitionBy("event_date")
   .mode("append")
   .saveAsTable("silver.events"))

maxRecordsPerFile limits file size but doesn't add parallelism; the salt does both, at the cost of up to four files per date.

And the bigger question: should the table be partitioned at all? Databricks' guidance is that tables under roughly 1 TB usually don't need directory partitioning, and that each partition should hold at least about 1 GB. Partitioning by a high-cardinality column like customer_id or by timestamp instead of date is how you end up with millions of tiny files. For new tables, liquid clustering is usually the better default.

Let Delta size the files for you

On Databricks, Delta has two features aimed at exactly this problem:

Optimized writes add a shuffle before the write so each task writes larger, well-sized files per table partition.
Auto compaction runs a small, synchronous compaction after a write when a partition has accumulated many small files.

ALTER TABLE silver.events SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true'
);

Or for a session:

spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

Recent Databricks Runtime versions turn these on by default for many operations (for example MERGE, UPDATE, and DELETE, and for Unity Catalog managed tables), so check your runtime's docs before adding manual repartition() calls that duplicate the work. Scheduled OPTIMIZE still handles whatever slips through.

How to check what you produced

After a write, look at the table, not your assumptions:

detail = spark.sql("DESCRIBE DETAIL silver.events").select("numFiles", "sizeInBytes").first()
print(detail.numFiles, round(detail.sizeInBytes / detail.numFiles / 1024**2, 1), "MB avg")

# Operation metrics from the last write
(spark.sql("DESCRIBE HISTORY silver.events LIMIT 1")
      .select("operation", "operationMetrics.numFiles", "operationMetrics.numOutputRows")
      .show(truncate=False))

An average file size in the low single-digit megabytes is a signal to revisit partitioning or turn on optimized writes. Something between roughly 100 MB and 1 GB is healthy for most analytic tables.

Interview questions worth practising

What's the difference between repartition() and coalesce(), and why can coalesce(1) make a job slower than repartition(1)?
A daily job writing with partitionBy("event_date") produces 15,000 files. Explain why and give two fixes.
Does repartition("customer_id") fix data skew? Why or why not?
What does AQE do to shuffle partitions, and what can't it do?
When would you not partition a Delta table at all?

If you can reason through those with DESCRIBE DETAIL output in front of you, you'll handle most write-performance questions.

PySpark Caching on Azure Databricks: cache() vs persist() vs checkpoint(), the Disk Cache, and When Caching Hurts

Karan Pulyala — Sun, 25 Oct 2026 14:22:08 +0000

"Should I cache this DataFrame?" is one of those questions that sounds simple in an interview and turns out to have a lot of layers. On Azure Databricks there are actually two different caches in play, and cache() is lazy, can silently serve stale data, and sometimes makes a job slower. This post walks through what each one does, when it helps, and how to prove it from the Spark UI.

Two caches, not one

On Databricks you'll run into:

The Spark cache (df.cache(), df.persist(), CACHE TABLE). It stores the result of a query plan in executor memory and/or local disk, in Spark's in-memory format. You control it explicitly.
The Databricks disk cache (it used to be called the Delta cache). It stores copies of remote Parquet/Delta files on the workers' local SSDs, so repeated reads skip cloud storage. It works automatically when enabled, and you don't call anything per DataFrame.

They solve different problems. The Spark cache saves recomputation (joins, aggregations, UDFs). The disk cache saves I/O from ADLS. Interviewers like asking for this distinction, so it's worth being crisp about it.

The Spark cache: lazy, plan-based, and per cluster

from pyspark.sql import functions as F

orders = spark.read.table("sales.orders")
customers = spark.read.table("sales.customers")

enriched = (
    orders.join(customers, "customer_id")
          .withColumn("order_month", F.date_trunc("month", "order_ts"))
          .filter(F.col("status") == "COMPLETE")
)

enriched.cache()      # nothing happens yet
enriched.count()      # this action materialises the cache

by_month   = enriched.groupBy("order_month").agg(F.sum("amount").alias("revenue"))
by_country = enriched.groupBy("country").agg(F.countDistinct("customer_id").alias("buyers"))

by_month.write.mode("overwrite").saveAsTable("reporting.revenue_by_month")
by_country.write.mode("overwrite").saveAsTable("reporting.buyers_by_country")

enriched.unpersist()

A few things to notice:

cache() is a transformation-like marker, not an action. Nothing is stored until an action runs. If you call cache() and then only use the DataFrame once, you paid the cost of writing the cache for nothing.
The cache is keyed on the logical plan. Any later query whose plan matches (or contains) the cached plan will read from cache. A DataFrame built with the same code in another notebook attached to the same cluster can hit the cache too, because the cache lives in the cluster's executors, not in your Python variable.
count() is a common way to force materialisation, but be aware that for some plans Spark can satisfy count() without computing every column. If you need the full thing materialised, run the action you actually care about, or use the noop writer, which computes every column and writes nothing:

# Forces full computation of every column, writes nothing
enriched.write.format("noop").mode("overwrite").save()

cache() vs persist() and storage levels

cache() is just persist() with the default storage level. For DataFrames that default is memory first, spilling to local disk when memory runs out (MEMORY_AND_DISK style). You can choose explicitly:

from pyspark import StorageLevel

big_lookup = spark.read.table("ref.product_attributes")

big_lookup.persist(StorageLevel.DISK_ONLY)       # keep executor memory free
# big_lookup.persist(StorageLevel.MEMORY_ONLY)   # recompute partitions that don't fit
# big_lookup.persist(StorageLevel.MEMORY_AND_DISK_2)  # replicated on two executors

print(big_lookup.storageLevel)

How I think about the choice:

MEMORY_AND_DISK (default): fine for most reuse cases.
MEMORY_ONLY: partitions that don't fit are recomputed on demand. Good when recomputation is cheap and memory is tight; bad when the lineage is expensive.
DISK_ONLY: useful when the result is large and executor memory is needed for shuffles and joins. Local SSD reads are still far cheaper than recomputing a multi-way join.
_2 variants: replicate each partition to a second executor. Rarely worth it unless you're on spot instances and lose executors often.

checkpoint() and localCheckpoint(): cutting the lineage

Caching keeps the lineage. If an executor dies, Spark recomputes the lost partitions from the original plan. In iterative jobs (loops that keep transforming the same DataFrame), the plan itself can get enormous, and planning time starts to dominate. That's where checkpointing helps:

spark.sparkContext.setCheckpointDir("abfss://scratch@mystorage.dfs.core.windows.net/checkpoints/")

df = spark.read.table("graph.nodes")
for i in range(20):
    df = df.withColumn("score", F.col("score") * 0.85 + 0.15).withColumn("iteration", F.lit(i))
    if i % 5 == 4:
        df = df.checkpoint()   # eager by default: writes files and truncates the plan

checkpoint() writes the data to reliable storage and returns a DataFrame whose plan is just "read these files". It survives executor loss.
localCheckpoint() stores data on executors instead. It's faster, but if an executor is lost the data is gone and the job fails rather than recomputes.
Note that checkpoint() returns a new DataFrame. Calling df.checkpoint() without reassigning does the work and throws the result away.

In practice, for a pipeline step that's reused across jobs, writing an intermediate Delta table is often clearer than either: it's inspectable, survives cluster restarts, and you can OPTIMIZE it.

The Databricks disk cache

The disk cache is controlled by a Spark config and is enabled by default on many worker types that have local SSDs. You can check and set it per cluster or per session:

print(spark.conf.get("spark.databricks.io.cache.enabled"))
spark.conf.set("spark.databricks.io.cache.enabled", "true")

It caches the remote data files that queries read, so the second query over the same columns and files is served from local disk. Because it caches files, it's automatically consistent with Delta: when a new version writes new files, those are fetched; old cached files simply stop being referenced.

You can warm it ahead of a dashboard refresh with SQL:

CACHE SELECT customer_id, amount, order_ts
FROM sales.orders
WHERE order_ts >= current_date() - INTERVAL 30 DAYS;

For BI-style workloads that hit the same tables repeatedly, the disk cache usually gives you most of the benefit with none of the bookkeeping of the Spark cache. Also note that serverless compute has its own rules around caching APIs, so check the current limitations for your workspace before relying on cache() there.

When caching makes things worse

This is the part interviewers really want to hear:

Single use. Caching something read once adds a write to memory/disk with no payoff.
Caching too early. Cache after filters and column pruning, not the raw table. spark.read.table("big").cache() stores every column, then your later select can't push down into the cached data.
Memory pressure. Cached blocks compete with execution memory. A large cache can push shuffles and joins into spilling, making the whole job slower. Watch for spill in the stage metrics after adding a cache.
Stale results. The Spark cache is a snapshot. If another job updates the underlying Delta table, a long-running notebook can keep serving the old cached data until you unpersist() or run REFRESH TABLE. The disk cache doesn't have this problem because it caches immutable files.
Forgetting to clean up. On shared all-purpose clusters, leftover caches eat memory for everyone. Use unpersist() when you're done, or spark.catalog.clearCache() at the end of a notebook.

Proving the cache is used

Don't guess; check:

enriched.explain()

Look for InMemoryTableScan (and InMemoryRelation) in the physical plan. If you see the original Scan parquet/FileScan and joins instead, your later query isn't matching the cached plan, often because it was built from a different DataFrame lineage.

In the Spark UI:

The Storage tab shows each cached DataFrame, its storage level, the fraction cached, and how much is in memory vs disk. "Fraction cached" below 100% with MEMORY_ONLY means partitions are being recomputed.
In the stage details, cached reads show up as fast tasks with no input from cloud storage.
For the disk cache, the query profile and the cluster's metrics show cache hit rates and bytes read from cache vs remote.

Interview questions worth practising

What's the difference between the Spark cache and the Databricks disk cache, and when would you use each?
You called df.cache() but the job didn't get faster. Name three possible reasons.
When would you choose checkpoint() over cache()? What about localCheckpoint()?
A notebook shows yesterday's numbers even though the table was updated an hour ago. What's a likely cause?
How do you confirm from the plan that a query is reading from the cache?

If you can answer those with the plan and the Storage tab in front of you, you've got caching covered.

I've put together a free 10-question sample of Azure Databricks and PySpark interview questions with worked answers, if you'd like more practice like this: free sample. The full set is in the Azure Databricks & PySpark Interview Pack.

PySpark Window Functions on Databricks: Latest-Record Dedup, Running Totals, Sessionisation, and the Frame Traps

Karan Pulyala — Sat, 24 Oct 2026 14:18:21 +0000

Window functions show up in almost every PySpark interview, and in almost every real pipeline: deduplicating CDC feeds, picking the latest record per key, running totals, sessionisation. They're also an easy place to write code that's correct on a sample and painfully slow (or subtly wrong) on production data.

This post walks through the patterns I use most, with code you can paste into a Databricks notebook, and the gotchas interviewers like to probe.

The setup

from pyspark.sql import functions as F
from pyspark.sql.window import Window

events = spark.createDataFrame(
    [
        ("c1", "2026-10-01 09:00:00", "view",     0.0),
        ("c1", "2026-10-01 09:05:00", "cart",     0.0),
        ("c1", "2026-10-01 09:07:00", "purchase", 40.0),
        ("c1", "2026-10-01 11:30:00", "view",     0.0),
        ("c2", "2026-10-01 10:00:00", "view",     0.0),
        ("c2", "2026-10-01 10:00:00", "view",     0.0),   # exact duplicate
        ("c2", "2026-10-02 08:15:00", "purchase", 15.0),
    ],
    "customer_id string, event_ts string, event_type string, amount double",
).withColumn("event_ts", F.to_timestamp("event_ts"))

A window spec has three parts: partitionBy (which rows belong together), orderBy (their order inside the group), and an optional frame (rowsBetween / rangeBetween, which rows around the current one the function can see).

Pattern 1: latest record per key (and why `row_number` beats `rank` here)

w_latest = Window.partitionBy("customer_id").orderBy(F.col("event_ts").desc())

latest = (
    events
    .withColumn("rn", F.row_number().over(w_latest))
    .filter("rn = 1")
    .drop("rn")
)

The interview question is usually: "What's the difference between row_number, rank, and dense_rank?"

row_number always gives 1, 2, 3, and so on, even for ties, so you get exactly one row per key.
rank gives tied rows the same number and leaves gaps (1, 1, 3).
dense_rank gives ties the same number with no gaps (1, 1, 2).

If two rows tie on event_ts, rank() == 1 returns both and your "latest" table quietly has duplicates. But row_number has its own trap: with ties, which row wins is non-deterministic and can change between runs. Always add a tiebreaker column so the result is reproducible:

w_latest = (
    Window.partitionBy("customer_id")
    .orderBy(F.col("event_ts").desc(), F.col("event_type").desc())
)

In a CDC feed, the tiebreaker is usually a sequence number or commit version, not a business column.

Pattern 2: dedup vs `dropDuplicates`

For exact duplicates, dropDuplicates(["customer_id", "event_ts", "event_type"]) is simpler and usually cheaper. Reach for row_number when you need to control which duplicate survives (newest ingested_at, highest version). Interviewers like asking which row dropDuplicates keeps. The honest answer is "an arbitrary one", so don't rely on it when the choice matters.

Pattern 3: running totals and the default frame trap

w_running = (
    Window.partitionBy("customer_id")
    .orderBy("event_ts")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

running = events.withColumn("running_spend", F.sum("amount").over(w_running))

Why spell out rowsBetween? When a window has an orderBy but no explicit frame, Spark defaults to RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW. With RANGE, all rows that share the same event_ts are treated as peers and included together. For customer c2, the two duplicate 10:00 rows would both show the combined total instead of stepping up one at a time. ROWS counts physical rows, which is what most people actually mean by "running total".

The same default bites last():

# Looks right, but returns the current row's value with the default frame
F.last("event_type").over(Window.partitionBy("customer_id").orderBy("event_ts"))

# What you probably wanted: last value in the whole partition
w_all = (
    Window.partitionBy("customer_id")
    .orderBy("event_ts")
    .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
)
F.last("event_type").over(w_all)

Pattern 4: time-based rolling windows with `rangeBetween`

"Spend in the last 24 hours" is a time range, not a row count. rangeBetween needs a numeric order column, so cast the timestamp to seconds:

w_24h = (
    Window.partitionBy("customer_id")
    .orderBy(F.col("event_ts").cast("long"))
    .rangeBetween(-24 * 3600, 0)
)

rolling = events.withColumn("spend_24h", F.sum("amount").over(w_24h))

rowsBetween(-6, 0) would give you "the last seven events", which is a different question. Mixing those up is a classic interview mistake.

Pattern 5: `lag`/`lead` and sessionisation (gaps and islands)

Split each customer's events into sessions whenever there's a gap of more than 30 minutes:

w_order = Window.partitionBy("customer_id").orderBy("event_ts")

sessions = (
    events
    .withColumn("prev_ts", F.lag("event_ts").over(w_order))
    .withColumn(
        "gap_min",
        (F.col("event_ts").cast("long") - F.col("prev_ts").cast("long")) / 60,
    )
    .withColumn(
        "new_session",
        F.when(F.col("prev_ts").isNull() | (F.col("gap_min") > 30), 1).otherwise(0),
    )
    .withColumn(
        "session_no",
        F.sum("new_session").over(
            w_order.rowsBetween(Window.unboundedPreceding, Window.currentRow)
        ),
    )
    .withColumn("session_id", F.concat_ws("-", "customer_id", "session_no"))
)

The trick is the "flag, then cumulative sum" pattern: mark the first row of each island with 1, and a running sum of those flags gives every row its island number. The same idea solves "consecutive login days", "status change periods", and most other gaps-and-islands questions.

For customer c1, the 09:00, 09:05, and 09:07 events land in session 1 and the 11:30 view starts session 2.

Pattern 6: compare a row with its group without a join

w_cust = Window.partitionBy("customer_id")

with_share = (
    events
    .withColumn("total_spend", F.sum("amount").over(w_cust))
    .withColumn(
        "pct_of_customer",
        F.when(F.col("total_spend") > 0, F.col("amount") / F.col("total_spend")),
    )
)

This replaces the "aggregate, then join back" pattern. It's usually cleaner, and it saves a second scan, but it still shuffles by customer_id.

Performance: what windows really cost

Every distinct partitionBy key set needs a shuffle and a sort. A few practical rules:

Reuse the same window spec. Several functions over identical partitionBy/orderBy are evaluated in one Window operator. Five slightly different specs can mean several shuffles. Check with df.explain() and count the Exchange and Window nodes.
Never use a window without partitionBy on big data. Window.orderBy("event_ts") alone moves every row into a single partition, and Spark logs a warning to tell you so. A global row_number over a billion rows is one task doing all the work.
Watch for skew. A single hot key (a bot account, a null customer ID) becomes one giant sorted partition. AQE's skew join handling helps joins, not windows, so you have to fix this yourself, for example by filtering or handling nulls separately before the window.
For top-1 per key, consider an aggregate. max_by (Spark 3.3+) avoids the sort:

latest_type = events.groupBy("customer_id").agg(
    F.max_by("event_type", "event_ts").alias("latest_event_type")
)

Newer Spark versions can also optimise a row_number() ... = 1 filter with a group limit, but max_by is a clean answer when you only need a few columns.

Interview questions worth practising

Your "latest record per key" job returns duplicates on some days. What would you check first?
Why might a running total show the same value on two adjacent rows?
What's the difference between rowsBetween(-6, 0) and rangeBetween(-6 * 86400, 0)?
How would you find customers who logged in on three or more consecutive days?
A window job has one task that runs for 40 minutes while the rest finish in seconds. What's going on and how would you fix it?

If you can answer those with code, window functions won't trip you up.

Spark Join Strategies on Databricks: Broadcast, Sort-Merge, Shuffle Hash, and What AQE Changes

Karan Pulyala — Fri, 23 Oct 2026 14:21:04 +0000

Most slow PySpark jobs I've debugged weren't slow because of bad code. They were slow because Spark picked a join strategy that made sense for the statistics it had, and those statistics were wrong. If you can read a physical plan and know why Spark chose a broadcast hash join over a sort-merge join, you can fix a big share of performance problems in minutes. It's also one of the most common senior data engineering interview topics.

This post walks through the join strategies Spark actually uses, how it chooses between them, how Adaptive Query Execution (AQE) changes that decision at runtime, and the hints you can use when you know better than the optimizer.

The strategies you'll see in a plan

For equi-joins (joins on a.key = b.key), Spark has three main physical strategies:

Strategy	Shuffle?	Sort?	Good when
Broadcast hash join (BHJ)	No (small side is broadcast)	No	One side is small enough to fit in memory on every executor
Shuffle hash join (SHJ)	Yes, both sides	No	Both sides are large, but each partition of the build side fits in memory
Sort-merge join (SMJ)	Yes, both sides	Yes	Both sides are large; the safe default

For non-equi joins (for example a.ts BETWEEN b.start AND b.end), Spark falls back to a broadcast nested loop join or a Cartesian product. Those are the ones that quietly turn a 5-minute job into a 5-hour job.

How to see which one you got

Always check the plan before tuning anything:

from pyspark.sql import functions as F

orders = spark.table("sales.orders")          # large fact table
customers = spark.table("sales.customers")    # dimension table

joined = orders.join(customers, "customer_id", "inner")
joined.explain("formatted")

In the output, look for one of BroadcastHashJoin, SortMergeJoin, or ShuffledHashJoin. A BroadcastExchange node means one side is being shipped to every executor. An Exchange hashpartitioning(customer_id, ...) on both sides means a full shuffle.

With AQE on (the default on Databricks), the plan you see before execution starts with AdaptiveSparkPlan isFinalPlan=false. The final plan is only known after the job runs, so check the SQL tab in the Spark UI to see what actually executed.

How Spark chooses

At planning time, for an equi-join, Spark roughly follows this order:

If there's a join hint, honour it (when it's valid for the join type).
If one side's estimated size is below spark.sql.autoBroadcastJoinThreshold (10 MB by default) and that side can be the build side for this join type, use a broadcast hash join.
If spark.sql.join.preferSortMergeJoin is false and the build side is small enough per partition, consider a shuffle hash join.
Otherwise, use a sort-merge join (when the keys are sortable).

The key word in step 2 is estimated. Spark estimates size from table statistics or file sizes, and after filters, aggregations, or UDFs those estimates can be badly off in either direction. That's why AQE matters.

Which side can be broadcast

This is a classic interview trap. The broadcast side is the build side, and not every join type allows every side to be the build side:

inner: either side can be broadcast.
left, left_semi, left_anti: only the right side can be broadcast.
right: only the left side can be broadcast.
full_outer: a broadcast hash join isn't possible.

So if you write small_df.join(big_df, "id", "left") and expect the small table to be broadcast, it won't be. Spark has to keep every row of the left side, so it can't build a hash table from it. Flip the join around or change the join type.

AQE: fixing the plan at runtime

With spark.sql.adaptive.enabled=true, Spark re-plans after each shuffle stage finishes, using real sizes instead of estimates. For joins, two things matter most:

Sort-merge join to broadcast join. If the actual shuffled size of one side ends up below spark.sql.adaptive.autoBroadcastJoinThreshold (which defaults to the same value as spark.sql.autoBroadcastJoinThreshold), AQE switches the join to a broadcast hash join. The shuffle already happened, but AQE can use a local shuffle reader to avoid a second network read and skip the sort.

Sort-merge join to shuffle hash join. If every post-shuffle partition is smaller than spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold, AQE can use a shuffle hash join instead, skipping the sort. That setting defaults to 0 (off) in open-source Spark, so it only kicks in if you set it.

A typical pattern: a filter cuts a 2 GB dimension down to 4 MB, the optimizer can't tell, plans a sort-merge join, and AQE saves you by converting it to a broadcast join at runtime.

spark.conf.set("spark.sql.adaptive.enabled", "true")

active = (spark.table("sales.customers")
          .filter(F.col("status") == "ACTIVE")
          .filter(F.col("country") == "CA"))

result = spark.table("sales.orders").join(active, "customer_id")
result.write.mode("overwrite").saveAsTable("sales.orders_active_ca")
# Spark UI > SQL tab: the final plan may show BroadcastHashJoin
# even though the initial plan showed SortMergeJoin.

Join hints: when you know better

Hints override the size-based choice. In PySpark:

from pyspark.sql.functions import broadcast

# Force a broadcast of the dimension table
orders.join(broadcast(customers), "customer_id")

# Same thing with the generic hint API
orders.join(customers.hint("broadcast"), "customer_id")

# Ask for a shuffle hash join or a sort-merge join explicitly
orders.join(customers.hint("shuffle_hash"), "customer_id")
orders.join(customers.hint("merge"), "customer_id")

And in SQL:

SELECT /*+ BROADCAST(c) */ o.*, c.segment
FROM sales.orders o
JOIN sales.customers c ON o.customer_id = c.customer_id;

If you put conflicting hints on both sides, Spark prioritises them in this order: BROADCAST, then MERGE, then SHUFFLE_HASH, then SHUFFLE_REPLICATE_NL. Hints that aren't valid for the join type (for example broadcasting the left side of a left outer join) are ignored, and Spark logs a warning.

When broadcasting goes wrong

Broadcasting isn't free. The driver collects the small side, builds the hash relation, and sends it to every executor. Common failure modes:

Driver out of memory. A "small" table that's actually 1.5 GB after decompression can take down the driver. Raising the threshold to something like 1 GB because it made one job faster is a classic way to break another job.
Broadcast timeout. If building the broadcast side involves heavy upstream work, you can hit spark.sql.broadcastTimeout (300 seconds by default). Caching or materialising the small side first usually helps more than raising the timeout.
Hard size limit. Spark refuses to broadcast tables over 8 GB, regardless of hints.
Executor memory pressure. Every executor holds a copy. A 500 MB broadcast across 200 executors is 100 GB of cluster memory.

A safe pattern is to compute the small side explicitly and check its size before forcing the hint:

dim = spark.table("sales.customers").select("customer_id", "segment")
dim_bytes = (spark.sql("DESCRIBE DETAIL sales.customers")
             .select("sizeInBytes").first()[0])  # Delta table size on disk

if dim_bytes < 200 * 1024 * 1024:   # 200 MB on disk, adjust for your cluster
    joined = orders.join(broadcast(dim), "customer_id")
else:
    joined = orders.join(dim, "customer_id")

Size on disk is compressed, so leave plenty of headroom; Parquet often expands several times in memory.

Keeping statistics honest

The optimizer can only choose well if it knows table sizes. On Databricks, run this on dimension tables that feed important joins:

ANALYZE TABLE sales.customers COMPUTE STATISTICS FOR ALL COLUMNS;

Column statistics also help the cost-based optimizer (spark.sql.cbo.enabled) reorder multi-way joins so the most selective joins happen first.

Interview questions worth practising

Why can't Spark broadcast the left side of a left outer join?
Your job shows SortMergeJoin in explain() but BroadcastHashJoin in the Spark UI. What happened?
When would you prefer a shuffle hash join to a sort-merge join?
Setting autoBroadcastJoinThreshold to 2 GB made one job faster and broke another. Why?
A join on BETWEEN conditions is very slow. What's Spark doing, and how would you rewrite it? (Hint: bucket the range into discrete keys so it becomes an equi-join, then filter.)

If you can answer those clearly, with a plan in front of you, you'll handle most join questions in a Databricks interview.

Delta Lake Table Maintenance on Databricks: OPTIMIZE, Z-ORDER vs Liquid Clustering, and VACUUM

Karan Pulyala — Thu, 22 Oct 2026 14:22:47 +0000

Most Delta tables don't get slow because of one bad query. They get slow because nobody looks after them. Streaming jobs and frequent small batch writes leave thousands of tiny files behind, deletes and updates leave old files sitting in storage, and the data layout drifts away from how people actually filter it.

This post covers the three maintenance tools you'll use on Azure Databricks, which also come up a lot in data engineering interviews: OPTIMIZE (compaction), data layout with Z-ORDER or liquid clustering, and VACUUM. For each one I'll explain what it does, show code, and point out the trap that catches people.

1. First, measure: is this table actually unhealthy?

Don't run maintenance blindly. DESCRIBE DETAIL tells you how many files a table has and how big it is:

detail = spark.sql("DESCRIBE DETAIL main.sales.orders").select(
    "numFiles", "sizeInBytes", "partitionColumns", "clusteringColumns"
).first()

avg_mb = detail.sizeInBytes / max(detail.numFiles, 1) / 1024 / 1024
print(f"files={detail.numFiles}, avg file size={avg_mb:.1f} MB")

If a table of a few hundred GB has tens of thousands of files averaging a few MB each, you have a small-file problem. Every file means extra listing, opening, and footer-reading work for each query, and Spark ends up scheduling far more tasks than it needs.

DESCRIBE HISTORY is the other quick check. It shows which operations write to the table and how often:

(spark.sql("DESCRIBE HISTORY main.sales.orders")
     .select("version", "timestamp", "operation", "operationMetrics")
     .show(10, truncate=False))

A wall of STREAMING UPDATE or WRITE entries every minute, each adding a handful of files, is the classic cause.

2. OPTIMIZE: compacting small files

OPTIMIZE rewrites small files into larger ones. It's a transactional operation, so readers keep seeing a consistent snapshot while it runs.

# SQL
spark.sql("OPTIMIZE main.sales.orders")

# Only compact recent partitions on a partitioned table
spark.sql("OPTIMIZE main.sales.orders WHERE order_date >= '2026-10-01'")

The same thing with the Python DeltaTable API:

from delta.tables import DeltaTable

dt = DeltaTable.forName(spark, "main.sales.orders")
metrics = dt.optimize().executeCompaction()
metrics.select("metrics.numFilesAdded", "metrics.numFilesRemoved").show(truncate=False)

Things worth knowing:

The WHERE clause only works on partition columns for a partitioned table. It's how you avoid recompacting years of cold data every night.
OPTIMIZE is idempotent. Running it again on an already-compacted table does very little work, so scheduling it daily is cheap once the table is in good shape.
Prevent the problem at write time too. Optimized writes and auto compaction reduce how many small files get created in the first place:

ALTER TABLE main.sales.orders SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true'
);

On Unity Catalog managed tables, predictive optimization can run OPTIMIZE and VACUUM for you. It's worth checking whether it's enabled before you build your own maintenance jobs.

3. Data layout: Z-ORDER vs liquid clustering

Compaction fixes file count. Layout fixes which rows live together, and that's what makes data skipping work. Delta stores min/max statistics per file; if rows for customer_id = 42 are spread across every file, no file can be skipped.

Z-ORDER (the older approach)

OPTIMIZE main.sales.orders ZORDER BY (customer_id, product_id);

Z-ORDER co-locates related values across multiple columns. Its weaknesses:

It isn't incremental. Re-running it rewrites a lot of data, which gets expensive on big tables.
The columns aren't stored with the table. If someone forgets the ZORDER BY clause in the next job, the layout quietly degrades.
It stacks on top of partitioning, so you still have to choose partition columns up front, and get them right.

Liquid clustering (the current recommendation)

Liquid clustering replaces both partitioning and Z-ORDER for new tables. You declare clustering keys on the table itself:

CREATE TABLE main.sales.orders (
  order_id     BIGINT,
  customer_id  BIGINT,
  product_id   BIGINT,
  order_date   DATE,
  amount       DECIMAL(12, 2)
)
CLUSTER BY (customer_id, order_date);

Or from PySpark:

(df.write
   .format("delta")
   .clusterBy("customer_id", "order_date")
   .saveAsTable("main.sales.orders"))

Then a plain OPTIMIZE clusters the data, incrementally, touching only data that isn't clustered yet:

OPTIMIZE main.sales.orders;

Why interviewers like this topic:

You can change the keys without rewriting the table. ALTER TABLE main.sales.orders CLUSTER BY (product_id) changes the keys used for future clustering. Existing data is left alone until you run OPTIMIZE ... FULL (on recent runtimes) to recluster everything.
No high-cardinality partition explosion. Clustering on customer_id is fine; partitioning on it would create millions of directories.
It can't be combined with partitioning or Z-ORDER on the same table. That's a common gotcha when migrating.
Keep the key count small. Databricks allows up to four clustering columns, and fewer keys usually work better. Pick the columns that actually appear in filters.

On recent runtimes with predictive optimization you can use CLUSTER BY AUTO and let Databricks pick keys from your query history.

How to choose keys: look at the WHERE clauses and join keys in your heaviest queries. Columns used in equality and range filters are good candidates. A column nobody filters on is wasted effort, no matter how "important" it seems.

4. VACUUM: reclaiming storage without breaking time travel

Delta never deletes data files during an update, delete, merge, or OPTIMIZE. It marks them as removed in the transaction log, and they stay in storage. That's what makes time travel work, and it's also why storage costs creep up.

VACUUM physically deletes files that are no longer referenced by any table version within the retention window (7 days by default).

-- See what would be deleted first
VACUUM main.sales.orders DRY RUN;

-- Delete files older than the default retention (7 days)
VACUUM main.sales.orders;

-- Explicit retention in hours
VACUUM main.sales.orders RETAIN 168 HOURS;

from delta.tables import DeltaTable

DeltaTable.forName(spark, "main.sales.orders").vacuum(168)  # hours

The traps:

VACUUM limits time travel. After vacuuming with a 7-day retention, SELECT * FROM t VERSION AS OF <older version> fails if that version needs deleted files. If the business needs 30 days of time travel, set the retention to match:

   ALTER TABLE main.sales.orders SET TBLPROPERTIES (
     'delta.deletedFileRetentionDuration' = 'interval 30 days',
     'delta.logRetentionDuration'         = 'interval 30 days'
   );

Don't disable the safety check casually. Setting spark.databricks.delta.retentionDurationCheck.enabled = false to run RETAIN 0 HOURS can delete files that a long-running query or a lagging streaming job still needs. That's how people corrupt readers in production.
VACUUM doesn't shrink the transaction log. Log cleanup is controlled separately by delta.logRetentionDuration (30 days by default) and happens automatically at checkpoints.
Deletion vectors change the picture slightly. With deletion vectors enabled, a DELETE or UPDATE can mark rows as deleted without rewriting the file. The old rows physically stay until the file is rewritten. If you need them gone for compliance, run REORG TABLE main.sales.orders APPLY (PURGE) and then VACUUM.

5. A simple maintenance job

If you aren't using predictive optimization, a nightly job like this covers most tables:

from delta.tables import DeltaTable

TABLES = ["main.sales.orders", "main.sales.order_items"]

for name in TABLES:
    dt = DeltaTable.forName(spark, name)

    before = spark.sql(f"DESCRIBE DETAIL {name}").first().numFiles
    dt.optimize().executeCompaction()   # also clusters liquid-clustered tables
    after = spark.sql(f"DESCRIBE DETAIL {name}").first().numFiles

    dt.vacuum()                          # uses the table's retention setting
    print(f"{name}: files {before} -> {after}")

Run it on a job cluster, outside peak hours, and stagger it away from heavy writers to avoid concurrent-write conflicts.

Interview-ready summary

Small files: diagnose with DESCRIBE DETAIL, fix with OPTIMIZE, prevent with optimized writes and auto compaction.
Layout: prefer liquid clustering for new tables; Z-ORDER is the legacy option and isn't incremental.
VACUUM: deletes unreferenced files older than the retention window (7 days by default), and that window is also your time-travel limit.
Safety: never drop the retention check to clean up faster unless you're certain no reader or stream needs the old files.

If you're preparing for Databricks interviews, I put together a free 10-question sample with worked answers on topics like these: free sample. The full interview pack has more if you find it useful.

Auto Loader and Structured Streaming on Azure Databricks: Checkpoints, Schema Evolution, and Exactly-Once

Karan Pulyala — Wed, 21 Oct 2026 14:24:41 +0000

Almost every Azure Databricks project I've seen starts the same way: files land in ADLS Gen2, and something has to pick up only the new ones, reliably, without reprocessing yesterday's data. Auto Loader plus Structured Streaming is the standard answer, and it's also a favourite interview topic because the details (checkpoints, schema evolution, exactly-once) separate people who've run it in production from people who've only read the docs.

Here's a practical walkthrough with PySpark code you can drop into a notebook.

1. The minimal Auto Loader pipeline

from pyspark.sql import functions as F

source_path = "abfss://landing@mystorage.dfs.core.windows.net/orders/"
checkpoint  = "abfss://checkpoints@mystorage.dfs.core.windows.net/bronze_orders/"

raw = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint + "_schema")
    .option("cloudFiles.inferColumnTypes", "true")
    .load(source_path))

bronze = (raw
    .withColumn("_ingested_at", F.current_timestamp())
    .withColumn("_source_file", F.col("_metadata.file_path")))

(bronze.writeStream
    .option("checkpointLocation", checkpoint)
    .trigger(availableNow=True)
    .toTable("bronze.orders"))

A few things worth being able to explain:

cloudFiles is Auto Loader. It tracks which files it has already ingested (in RocksDB inside the checkpoint), so each file is processed exactly once, even if you have millions of them.
schemaLocation is where Auto Loader stores the inferred schema and its history. Without it you'd have to pass an explicit schema.
inferColumnTypes makes JSON/CSV columns come out as real types instead of all strings.
_metadata.file_path gives you lineage back to the source file for free. Very handy when someone asks "which file did this bad row come from?"

2. `availableNow` vs. continuous streaming

The trigger is a cost decision, not just a technical one.

# Process everything that's new, in multiple micro-batches, then stop
.trigger(availableNow=True)

# Keep running, start a micro-batch every 5 minutes
.trigger(processingTime="5 minutes")

availableNow=True is what I'd use for most batch-style ingestion: schedule the notebook as a Databricks job every hour, and the cluster only lives as long as there's work. It replaced the older trigger(once=True), which squeezed everything into a single batch and could blow up memory after a long backlog. availableNow respects rate limits, so you can cap each batch:

.option("cloudFiles.maxFilesPerTrigger", 1000)
.option("cloudFiles.maxBytesPerTrigger", "10g")

When both are set, a batch stops at whichever limit is hit first.

Interview answer in one line: same code, same checkpoint, so you can move from an hourly availableNow job to an always-on stream later without reprocessing anything.

3. Checkpoints: the rules people break

The checkpoint is the stream's memory: source offsets, which batches were committed, any state, and for Auto Loader the list of discovered files. Rules I'd want any teammate to know:

One checkpoint per stream. Two queries sharing a checkpoint will corrupt each other's progress.
Don't delete it to "fix" a problem unless you actually want to reprocess everything from the start of the source.
Treat it as tied to the query. Changing the source, or making big changes to aggregations/joins with state, can make an old checkpoint incompatible. Adding a filter or a new projected column is usually fine.
Keep it next to the data's lifecycle. If you drop and recreate the target table, you almost always want a new checkpoint too.

To see what Auto Loader has actually ingested, Databricks has a table-valued function you can query directly:

SELECT path, commit_time
FROM cloud_files_state('abfss://checkpoints@mystorage.dfs.core.windows.net/bronze_orders/')
ORDER BY commit_time DESC
LIMIT 20;

4. Schema evolution without 2 a.m. pages

Upstream teams add columns. The question is what your pipeline does when they do.

.option("cloudFiles.schemaEvolutionMode", "addNewColumns")  # default when you don't supply a schema

The modes:

Mode	What happens on a new column
`addNewColumns`	Stream fails once, schema is updated in `schemaLocation`, next run includes the column
`rescue`	Schema never changes; unexpected data goes into `_rescued_data`
`failOnNewColumns`	Stream fails and stays failed until you change the schema
`none`	New columns are ignored

The "fails once" behaviour of addNewColumns surprises people. It's by design: the stream stops, records the new schema, and a restart picks it up. In a Databricks job, set the task to retry on failure and it heals itself. On the write side, add .option("mergeSchema", "true") so the Delta table accepts the new column too.

I also like pinning the columns that matter with schema hints, so inference can't guess wrong:

.option("cloudFiles.schemaHints", "order_id BIGINT, amount DECIMAL(18,2), order_ts TIMESTAMP")

And always check _rescued_data. Auto Loader adds it by default when it infers the schema, and it captures values that didn't fit (a string in a numeric column, a field with different casing). A simple data quality check:

bad = spark.table("bronze.orders").filter("_rescued_data IS NOT NULL")
display(bad.select("_source_file", "_rescued_data").limit(50))

5. Exactly-once with `foreachBatch`

Writing straight to a Delta table with toTable is exactly-once out of the box: Delta and the checkpoint together make sure a batch that's retried doesn't get committed twice.

foreachBatch is where people lose that guarantee. It gives you full batch APIs (MERGE, writing to several tables), but if the job crashes after your write and before the checkpoint commit, the same batch runs again. Two ways to stay safe:

Option A: make the write idempotent with Delta's transaction options.

APP_ID = "bronze_to_silver_orders"  # stable, unique per stream

def write_batch(batch_df, batch_id):
    (batch_df.write
        .format("delta")
        .mode("append")
        .option("txnAppId", APP_ID)
        .option("txnVersion", batch_id)
        .saveAsTable("silver.orders_append"))

(spark.readStream.table("bronze.orders")
    .writeStream
    .foreachBatch(write_batch)
    .option("checkpointLocation", ".../checkpoints/silver_orders/")
    .trigger(availableNow=True)
    .start())

Delta remembers the highest txnVersion it committed for that txnAppId and silently skips a repeat. If you write to two tables in one batch, use the same pair of options on each write.

Option B: use MERGE and deduplicate inside the batch.

from delta.tables import DeltaTable
from pyspark.sql import Window

def upsert_batch(batch_df, batch_id):
    w = Window.partitionBy("order_id").orderBy(F.col("order_ts").desc())
    latest = (batch_df
        .withColumn("rn", F.row_number().over(w))
        .filter("rn = 1")
        .drop("rn"))

    (DeltaTable.forName(batch_df.sparkSession, "silver.orders").alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedUpdateAll(condition="s.order_ts > t.order_ts")
        .whenNotMatchedInsertAll()
        .execute())

Replaying this batch produces the same end state, so retries are harmless. The in-batch dedup matters because MERGE fails if several source rows match the same target row.

Note batch_df.sparkSession instead of the global spark: on newer runtimes and shared clusters that's the safe way to reach the session inside the function.

6. Directory listing vs. file notifications

By default Auto Loader discovers files by listing the input directory. That's fine for most workloads, and incremental listing keeps it cheap when file names are lexically ordered (like date-based paths).

For very high file volumes, file notification mode subscribes to storage events instead (Event Grid and a storage queue on Azure):

.option("cloudFiles.useNotifications", "true")

It needs permissions to create or use those Azure resources, so it's a conversation with your platform team, and newer Databricks workspaces also offer managed file events through Unity Catalog external locations. In an interview, knowing why you'd switch (listing cost and latency growing with directory size) matters more than the exact setup.

7. Monitoring a running stream

q = spark.streams.active[0]
print(q.status)
print(q.lastProgress["numInputRows"], q.lastProgress["batchDuration"])

For Auto Loader, lastProgress also includes a source metric for the backlog (numFilesOutstanding / numBytesOutstanding), which is the number I'd alert on: a growing backlog means ingestion isn't keeping up.

Quick checklist

cloudFiles + schemaLocation + a unique checkpoint per stream.
availableNow=True for scheduled jobs, with maxFilesPerTrigger/maxBytesPerTrigger caps.
addNewColumns plus job retries, mergeSchema on write, schema hints for key columns.
Watch _rescued_data.
In foreachBatch, use txnAppId/txnVersion or an idempotent MERGE.
Alert on backlog, not just job failure.

If you're preparing for a Databricks or PySpark interview, there's a free 10-question sample with worked answers in this style. The full Azure Databricks & PySpark Interview Pack has the rest if it helps.

Delta Lake MERGE in PySpark: Upserts, SCD Type 2, and the Traps Interviewers Ask About

Karan Pulyala — Tue, 20 Oct 2026 14:20:57 +0000

MERGE INTO is the workhorse of almost every Delta Lake pipeline on Azure Databricks. It's also one of the most common follow-up topics in data engineering interviews: once you say "I upsert into Delta," the next question is usually "OK, what happens when the source has duplicates?" or "How would you build SCD Type 2 with it?"

This post walks through the patterns I'd want a candidate (or a teammate) to know cold, with PySpark code you can run in a notebook.

1. The basic upsert

from delta.tables import DeltaTable
from pyspark.sql import functions as F

target = DeltaTable.forName(spark, "silver.customers")
updates = spark.table("bronze.customer_updates")

(target.alias("t")
    .merge(updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Delta joins source to target on the merge condition, finds the files that contain matched rows, rewrites them, and commits new files atomically. Readers never see a half-applied merge.

The SQL equivalent:

MERGE INTO silver.customers t
USING bronze.customer_updates s
ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

2. Trap #1: duplicate keys in the source

If two source rows match the same target row, the merge fails with an error along the lines of "multiple source rows matched and attempted to modify the same target row." Delta refuses to guess which one wins, which is the right call.

Interviewers love this one because the fix shows whether you think about data quality. Deduplicate the source first, keeping the latest record per key:

from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())

latest_updates = (updates
    .withColumn("rn", F.row_number().over(w))
    .filter("rn = 1")
    .drop("rn"))

Use row_number(), not rank(), so ties on updated_at still produce exactly one row. If ties are realistic, add a second ordering column (an ingestion timestamp or file name) so the result is deterministic.

3. Trap #2: rewriting rows that didn't change

whenMatchedUpdateAll() with no condition updates every matched row, even when nothing changed. In a daily full-snapshot feed where 98% of rows are identical, that means rewriting a lot of files for no reason.

Add a condition so only real changes are written. Use the null-safe equality operator <=>, because a <> b is NULL (not true) when either side is NULL:

change_cond = "NOT (t.email <=> s.email AND t.city <=> s.city AND t.tier <=> s.tier)"

(target.alias("t")
    .merge(latest_updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll(condition=change_cond)
    .whenNotMatchedInsertAll()
    .execute())

For wide tables, comparing a hash of the business columns is easier to maintain:

cols = ["email", "city", "tier"]
hashed = latest_updates.withColumn(
    "row_hash", F.sha2(F.concat_ws("||", *[F.coalesce(F.col(c).cast("string"), F.lit("<null>")) for c in cols]), 256)
)
# store row_hash in the target too, then: condition="t.row_hash <> s.row_hash"

4. Trap #3: scanning the whole target

The merge condition decides how much of the target Delta has to read. If the target is partitioned by event_date and your source only touches the last few days, say so in the condition:

(target.alias("t")
    .merge(
        latest_updates.alias("s"),
        "t.event_date >= date_sub(current_date(), 3) AND t.event_id = s.event_id AND t.event_date = s.event_date"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

A literal-style predicate on the partition (or clustering / Z-order) column lets Delta prune files before the join. Without it, every merge reads the full table, and your job gets slower every day as the table grows. This is one of the most common "why is my merge getting slower?" root causes I've seen.

Two related knobs worth mentioning in an interview:

If the source is small, Spark will usually broadcast it. Check the plan in the Spark UI rather than assuming.
On newer Databricks runtimes, deletion vectors let Delta mark rows as deleted/updated without rewriting whole files immediately, which can make merges with few changed rows much cheaper. Run OPTIMIZE regularly so the table doesn't accumulate small files.

5. Handling deletes from the source system

If the source is a full snapshot and a key disappears, you may want to delete (or soft-delete) it in the target. Delta Lake 2.3+ / recent Databricks runtimes support whenNotMatchedBySource:

(target.alias("t")
    .merge(latest_updates.alias("s"), "t.customer_id = s.customer_id")
    .whenMatchedUpdateAll(condition=change_cond)
    .whenNotMatchedInsertAll()
    .whenNotMatchedBySourceUpdate(set={"is_active": F.lit(False)})
    .execute())

Only do this when the source really is a complete snapshot. Running it against an incremental feed will "deactivate" every row that just didn't change today.

6. SCD Type 2 with a single MERGE

The classic question: keep full history, with is_current, start_date, and end_date columns. A changed customer needs two actions in one atomic commit: close the old row and insert a new current row. A single MERGE can't both update and insert for the same source row, so the trick is to stage the changed rows twice.

dim = DeltaTable.forName(spark, "gold.dim_customer")
src = latest_updates  # customer_id, email, city, tier, updated_at

current = dim.toDF().filter("is_current = true")

scd_change = "NOT (t.email <=> s.email AND t.city <=> s.city AND t.tier <=> s.tier)"

# Rows whose attributes changed: these need a NEW current row inserted
changed = (src.alias("s")
    .join(current.alias("t"), "customer_id")
    .where(scd_change)
    .select("s.*"))

staged = (
    # merge_key = NULL never matches, so these rows fall through to INSERT
    changed.withColumn("merge_key", F.lit(None).cast(src.schema["customer_id"].dataType))
    .unionByName(
        # merge_key = customer_id matches the current row, so we can close it out
        src.withColumn("merge_key", F.col("customer_id"))
    )
)

(dim.alias("t")
    .merge(staged.alias("s"), "t.customer_id = s.merge_key AND t.is_current = true")
    .whenMatchedUpdate(
        condition=scd_change,
        set={"is_current": F.lit(False), "end_date": F.col("s.updated_at")}
    )
    .whenNotMatchedInsert(values={
        "customer_id": "s.customer_id",
        "email": "s.email",
        "city": "s.city",
        "tier": "s.tier",
        "is_current": F.lit(True),
        "start_date": F.col("s.updated_at"),
        "end_date": F.lit(None).cast("timestamp"),
    })
    .execute())

What happens to each kind of source row:

Source row	`merge_key`	Result
Brand-new customer	its id	No match, inserted as current
Existing, unchanged	its id	Matches, but condition is false, so nothing happens
Existing, changed (copy 1)	its id	Matches, old row closed (`is_current = false`)
Existing, changed (copy 2)	NULL	No match, new current row inserted

Because it's one MERGE, the close-out and the insert land in the same Delta commit, so no reader ever sees a customer with zero or two current rows.

7. Verify what the merge actually did

Don't guess, check. Every merge records operation metrics in the table history:

(spark.sql("DESCRIBE HISTORY gold.dim_customer LIMIT 1")
    .select("version", "operation", "operationMetrics")
    .show(truncate=False))

Look at numTargetRowsUpdated, numTargetRowsInserted, numTargetFilesAdded, and numTargetFilesRemoved. If a merge that changed 200 rows removed 4,000 files, your condition isn't pruning and that's where to dig.

Quick interview checklist

Deduplicate the source before merging, and explain why Delta errors instead of picking a row.
Use null-safe comparisons (<=>) in change detection.
Add a condition to whenMatched so unchanged rows aren't rewritten.
Put partition or clustering predicates in the merge condition so files get pruned.
Know the staged-union trick for SCD Type 2 in a single atomic merge.
Read DESCRIBE HISTORY metrics to prove what happened.

If you're prepping for a Databricks or PySpark interview, I put together a free 10-question sample with worked answers in this style. The full Azure Databricks & PySpark Interview Pack has the rest if you find it useful.

Diagnosing and Fixing Data Skew in PySpark on Azure Databricks (2026 Edition)

Karan Pulyala — Mon, 19 Oct 2026 20:11:31 +0000

A familiar failure: the stage hits 99% and stalls. 199 tasks are done, one is still running, and adding nodes doesn't help, because the bottleneck is one partition, not the cluster.

That's data skew. In this post I'll go through how I diagnose it and how I fix it on Azure Databricks as of 2026, roughly in the order I'd try each fix. Every snippet runs on Databricks Free Edition, which is serverless-only, so I'll flag the places where serverless behaves differently.

Not affiliated with or endorsed by Databricks. Config names and defaults are from the Apache Spark and Databricks docs as of October 2026. Check them against your runtime version.

What skew actually is

A shuffle sends every row with the same key to the same partition. That's what makes joins and groupBy work. If one key holds 30% of your rows, one task gets 30% of the work, and the stage waits for it.

The usual culprits:

NULL or placeholder keys ('UNKNOWN', -1, ''). In my experience this is the cause more often than anything else.
A handful of "whale" entities: one huge customer, one default store, one bot user agent.
Time skew: one day (Black Friday, a backfill date) is 50Ã— the others.
File/read skew, which isn't about keys at all. A few huge or unsplittable files (gzipped CSV, for example) make the first stage lopsided.

Step 1: Build a skewed dataset to practice on

Paste this into a Free Edition notebook. It creates 20M orders: ~30% with a NULL customer, ~20% for one whale customer (42):

from pyspark.sql import functions as F

N = 20_000_000
orders = (
    spark.range(N)
    .withColumn("r", F.rand(seed=7))
    .withColumn(
        "customer_id",
        F.when(F.col("r") < 0.30, F.lit(None).cast("bigint"))   # ~30% NULL
         .when(F.col("r") < 0.50, F.lit(42).cast("bigint"))     # ~20% whale
         .otherwise((F.rand(seed=11) * 1_000_000).cast("bigint"))
    )
    .withColumn("amount", F.round(F.rand(seed=3) * 100, 2))
    .drop("r")
)

customers = (
    spark.range(1_000_000).withColumnRenamed("id", "customer_id")
    .withColumn("segment", F.when(F.col("customer_id") % 10 == 0, "enterprise").otherwise("smb"))
)

Step 2: Confirm it. Don't guess.

Key distribution. This is the cheapest check and the one I always run first:

total = orders.count()
(orders.groupBy("customer_id").count()
    .withColumn("pct_of_rows", F.round(100 * F.col("count") / total, 2))
    .orderBy(F.desc("count"))
    .show(10))

Double-digit percentages at the top mean you've found it.

Partition-level distribution. This shows what a shuffle on that key does to your tasks:

parts = (orders.repartition(200, "customer_id")
    .groupBy(F.spark_partition_id().alias("pid")).count())

parts.agg(
    F.max("count").alias("max_rows"),
    F.percentile_approx("count", 0.5).alias("median_rows"),
).withColumn("max_over_median", F.col("max_rows") / F.col("median_rows")).show()

A max/median ratio in the tens or hundreds means one task will do most of the work, no matter how big your cluster is.

In the Spark UI (classic compute). Open the slow stage and look at Summary Metrics. Compare Max with Median for Duration, Shuffle Read Size / Records, and Spill (Disk). A healthy stage has a fairly flat distribution. A skewed stage has a median of seconds and a max of minutes, and the Event Timeline shows one long bar.

On serverless (including Free Edition) there's no Spark UI. Click See performance under the cell and open the query profile. Look for the operator with a disproportionate share of time and rows. It's operator-level, not a per-task histogram, so the key-distribution query above is even more useful there.

Step 3: Let AQE do its job (and know when it can't)

Adaptive Query Execution re-plans at stage boundaries using real shuffle statistics. It's on by default in Spark 3.2+ and on Databricks. Its skew-join feature splits an oversized partition into smaller ones and replicates the matching partition on the other side. A partition counts as skewed only when both of these are true:

it's larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) Ã— the median partition size, and
it's larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB).

Split sizes aim at spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64MB). When it works, the final plan shows a SortMergeJoin with isSkew=true.

joined = orders.join(customers, "customer_id", "left")
joined.write.mode("overwrite").format("noop").save()   # run it without writing anything

noop runs the whole job without writing output. Then check the plan in the query profile (serverless) or the SQL tab (classic).

What AQE won't handle, according to the docs:

Broadcast joins. Skew handling only applies to sort-merge and shuffle-hash joins.
Some join sides. In a LEFT OUTER join only the left side can be split. FULL OUTER gets no skew handling at all.
Partitions under 256MB. If your "skew" is 200MB against a 2MB median, the threshold blocks it. On classic compute you can lower the thresholds. On serverless you can't, because only a short allowlist of Spark confs (like spark.sql.shuffle.partitions, which defaults to auto there) can be set.
Scan-stage skew, aggregations and window functions. The skew-join rule is about joins.

One honest caveat about the demo: on a small Free Edition dataset the skewed partition may not cross 256MB, so you may not see isSkew=true. That's the threshold working as designed, not a broken demo.

Step 4: Broadcast the small side

If the other side fits in memory, a broadcast hash join removes the shuffle completely. With no shuffle there are no hot partitions, so skew stops mattering:

joined = orders.join(F.broadcast(customers), "customer_id", "left")

Spark auto-broadcasts below spark.sql.autoBroadcastJoinThreshold (10MB in OSS). Databricks AQE can also switch to a broadcast at runtime below spark.databricks.adaptive.autoBroadcastJoinThreshold (documented at 30MB). Databricks' own guidance is still to hint the broadcast when you know a side is small, because a planned broadcast beats one that AQE only chooses after both sides have already shuffled.

Two cautions: compressed Parquet can expand a lot in memory, and you can't broadcast the preserved side of an outer join (the left side of a LEFT JOIN).

Step 5: Treat hot keys separately (especially NULLs)

For NULL keys this is free. NULLs never match in an equi-join, so pushing them through the shuffle is pure waste:

is_null = F.col("customer_id").isNull()

matched = orders.filter(~is_null).join(customers, "customer_id", "left")
nulls   = orders.filter(is_null)            # no partner rows to find anyway

result = matched.unionByName(nulls, allowMissingColumns=True)

For an inner join, just filter the NULLs out. The same split works for a known whale key: send it through a broadcast path and the rest through the normal join.

Step 6: Salting, the targeted version

Salting spreads one key across N partitions by adding a random suffix on the big side and replicating the small side N times. The common version salts every key, which multiplies the whole dimension table by N. I prefer to salt only the hot keys:

SALT = 16
hot_keys = [42]  # take these from your Step 2 profiling, not from a hard-coded guess

orders_s = orders.withColumn(
    "salt",
    F.when(F.col("customer_id").isin(hot_keys), (F.rand(seed=1) * SALT).cast("int"))
     .otherwise(F.lit(0))
)

customers_s = customers.withColumn(
    "salt",
    F.explode(
        F.when(F.col("customer_id").isin(hot_keys), F.sequence(F.lit(0), F.lit(SALT - 1)))
         .otherwise(F.array(F.lit(0)))
    )
)

salted = orders_s.join(customers_s, ["customer_id", "salt"], "left").drop("salt")

Only the whale's dimension row gets replicated (16 rows, not 16 million). Size SALT so each slice of the hot key is close to a normal partition.

For aggregations, be selective too. Spark already does partial aggregation on the map side, so plain sum/count on a skewed key often hurts less than you'd expect. Salting pays off for aggregations that can't be combined that way, like collect_list, exact countDistinct and percentiles, and for window functions partitioned by a hot key. Two-phase aggregation looks like this:

partial = (orders.withColumn("salt", (F.rand(seed=5) * SALT).cast("int"))
    .groupBy("customer_id", "salt")
    .agg(F.sum("amount").alias("s"), F.count("*").alias("c")))

final = partial.groupBy("customer_id").agg(F.sum("s").alias("total_amount"),
                                           F.sum("c").alias("order_count"))

Step 7: Repartitioning (and its trap)

repartition(400, "customer_id") does not fix key skew. Every row for key 42 still hashes to the same partition, and you've paid for a shuffle. What helps:

repartition(400) with no key (round-robin) when you just need even work for narrow, row-by-row processing.
The REBALANCE hint before a write. AQE can split skewed partitions in a rebalance (spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled, default true), which avoids one giant output file:

orders.hint("rebalance").write.mode("overwrite").saveAsTable("workspace.default.orders_demo")

(workspace.default is the default catalog and schema on Free Edition. Use your own elsewhere.)

Step 8: Fix the layout when the skew is on read

If the first stage is the lopsided one, AQE can't help, because there's no shuffle statistic to re-plan from yet. That's a file-layout problem: a few giant files, thousands of tiny ones, unsplittable compression, or Hive-style partitions where one date= folder is huge.

On Delta, the current recommendation for new tables is liquid clustering. Databricks lists "tables with heavy data skew" among its target use cases:

spark.sql("""
  CREATE OR REPLACE TABLE workspace.default.orders_clustered
  CLUSTER BY (customer_id)
  AS SELECT * FROM workspace.default.orders_demo
""")
spark.sql("OPTIMIZE workspace.default.orders_clustered")

Be clear about what this buys you: evenly sized files and better data skipping. It does not remove skew from a later join on customer_id. The shuffle still sends key 42 to one partition. With predictive optimization on Unity Catalog managed tables, CLUSTER BY AUTO lets Databricks choose the keys.

How to explain this in an interview

Interviewers are checking whether you have a process, not a definition. A structure that works:

Symptom: "A stage stuck at 99% with one long task. Max task duration far above the median, a big shuffle read on that task, often spill."
Confirm: "I profile the key distribution, and the first thing I check is NULL and default keys."
Fix in order of cost: "Check that AQE skew join is applicable (join type, thresholds), broadcast the small side if I can, route NULL or hot keys around the join, and use targeted salting as a last resort."
Know the limits: "AQE only handles skew in shuffle joins. It won't touch broadcast joins or scan-stage skew, and on serverless I can't tune its thresholds."
Close the loop: "Then I verify. I check for isSkew=true in the plan, compare max against median again, and add a skew check to the pipeline so a new whale customer doesn't bring it back."

Point 5 is where you stand out. Most candidates stop at "I'd use salting."

Checklist

[ ] Max vs. median task duration and shuffle read (Spark UI), or the slowest operator (query profile)
[ ] Top-10 key distribution, including NULL and placeholder values
[ ] Is it a join, an aggregation, a window function, or the scan stage?
[ ] AQE on and applicable: shuffle join, a join type that allows splitting that side, partitions over the thresholds
[ ] Small side broadcastable? Hint it.
[ ] NULLs filtered or routed around the join
[ ] Salt only the hot keys, sized from real numbers
[ ] Scan-stage skew â†’ fix the files (compaction, liquid clustering), not the join
[ ] Re-measure, then add monitoring

If you're prepping for Databricks interviews, I put 10 questions like this with model answers into a free sample: https://pulyala.gumroad.com/l/vonemy (the full 61-question pack is here: https://pulyala.gumroad.com/l/ivjsaf).

Not affiliated with Databricks. Databricks and Azure Databricks are trademarks of their respective owners.

DEV Community: Karan Pulyala

Delta Lake Change Data Feed on Azure Databricks: readChangeFeed, Incremental MERGE, and the Retention Traps

Turning it on

What a change row looks like

Batch reads

Streaming reads (usually the better choice)

Turning changes into a correct MERGE

Where pre-images earn their keep: aggregates

The traps interviewers ask about

Interview questions to practise

Spark JDBC Parallel Reads on Azure Databricks: partitionColumn, Bounds, Predicates, and Not Knocking Over Your Database

Why the default JDBC read uses one task

Parallel reads with partitionColumn

The skew problem: equal ranges, unequal rows

Don't knock over the source

fetchsize, pushdown, and reading only what you need

Writing back over JDBC

When not to use plain JDBC

Interview questions to practise

Delta Lake Schema Evolution and Time Travel on Azure Databricks: mergeSchema, overwriteSchema, Column Mapping, and RESTORE

Schema enforcement: Delta says no by default

Option 1: mergeSchema adds columns

Schema evolution in MERGE

Option 2: overwriteSchema replaces everything

Renaming and dropping columns: column mapping

Time travel: reading the past

RESTORE: undoing a bad write

Why time travel isn't a backup

Interview questions to practise

Repartition vs Coalesce vs partitionBy on Azure Databricks: Controlling Output Files When Writing Delta Tables

Three different things called "partition"

repartition(): a full shuffle, on purpose

coalesce(): cheap, but it can shrink your whole stage

Shuffle partitions and AQE

partitionBy(): where file explosions come from

Let Delta size the files for you

How to check what you produced

Interview questions worth practising

PySpark Caching on Azure Databricks: cache() vs persist() vs checkpoint(), the Disk Cache, and When Caching Hurts

Two caches, not one

The Spark cache: lazy, plan-based, and per cluster

cache() vs persist() and storage levels

checkpoint() and localCheckpoint(): cutting the lineage

The Databricks disk cache

When caching makes things worse

Proving the cache is used

Interview questions worth practising

PySpark Window Functions on Databricks: Latest-Record Dedup, Running Totals, Sessionisation, and the Frame Traps

The setup

Pattern 1: latest record per key (and why row_number beats rank here)

Pattern 2: dedup vs dropDuplicates

Pattern 3: running totals and the default frame trap

Pattern 4: time-based rolling windows with rangeBetween

Pattern 5: lag/lead and sessionisation (gaps and islands)

Pattern 6: compare a row with its group without a join

Performance: what windows really cost

Interview questions worth practising

Spark Join Strategies on Databricks: Broadcast, Sort-Merge, Shuffle Hash, and What AQE Changes

The strategies you'll see in a plan

How to see which one you got

How Spark chooses

Which side can be broadcast

AQE: fixing the plan at runtime

Join hints: when you know better

When broadcasting goes wrong

Keeping statistics honest

Interview questions worth practising

Delta Lake Table Maintenance on Databricks: OPTIMIZE, Z-ORDER vs Liquid Clustering, and VACUUM

1. First, measure: is this table actually unhealthy?

2. OPTIMIZE: compacting small files

3. Data layout: Z-ORDER vs liquid clustering

Z-ORDER (the older approach)

Liquid clustering (the current recommendation)

4. VACUUM: reclaiming storage without breaking time travel

5. A simple maintenance job

Interview-ready summary

Auto Loader and Structured Streaming on Azure Databricks: Checkpoints, Schema Evolution, and Exactly-Once

1. The minimal Auto Loader pipeline

2. availableNow vs. continuous streaming

3. Checkpoints: the rules people break

Pattern 1: latest record per key (and why `row_number` beats `rank` here)

Pattern 2: dedup vs `dropDuplicates`

Pattern 4: time-based rolling windows with `rangeBetween`

Pattern 5: `lag`/`lead` and sessionisation (gaps and islands)

2. `availableNow` vs. continuous streaming

5. Exactly-once with `foreachBatch`