PySpark interview questions for Data Engineer roles

An RDD (resilient distributed dataset) is Spark's original low-level abstraction: an immutable, partitioned collection of objects that Spark can process in parallel and rebuild from its lineage if a partition is lost. DataFrames and Datasets are built on top of it and add a schema, which lets Spark's optimizer plan the work.

Why DataFrames are the default

• Spark understands column operations, so it can push filters to the source, read only needed columns, choose join strategies and generate efficient code.
• In Python, RDD functions run in Python worker processes with data serialised back and forth, which is slow; DataFrame operations run in the JVM.
• In Python a DataFrame is the equivalent of a Dataset of rows; there is no typed Dataset API.

When an RDD still fits Low-level control over partitioning, or processing unstructured data that doesn't fit a table. Even then, most code converts to a DataFrame quickly. Spark Connect, used by serverless compute and many newer setups, doesn't offer the RDD API at all.

A job is the work Spark runs to complete one action, such as count(), collect() or a write. The job is split into stages at shuffle boundaries, and each stage is split into tasks, one per data partition, which run in parallel on executors.

Job

• Triggered by an action; transformations alone don't start one.
• One action can produce more than one job: reading a CSV with schema inference, or adaptive query execution re-planning, can add jobs.

Stage

• A set of operations that can run without moving data between executors: reading, filtering, select, withColumn and other narrow transformations are pipelined together.
• A wide transformation that needs a shuffle (groupBy, most joins, repartition, distinct, orderBy) ends one stage and starts the next. The earlier stage writes shuffle files; the next one reads them.
• Stages depend on each other, forming a DAG; independent stages can run at the same time.

Task

• The smallest unit of work: one stage's operations on one partition.
• A stage over 200 partitions has 200 tasks. Each task runs on one core of an executor.
• The number of tasks in a stage that reads files depends on file sizes and spark.sql.files.maxPartitionBytes (128 MB by default); after a shuffle it depends on spark.sql.shuffle.partitions and adaptive query execution.

Example

(spark.read.parquet("/data/sales")        # stage 1: read, filter, partial aggregation
    .filter("amount > 0")
    .groupBy("region")                     # shuffle boundary
    .sum("amount")                         # stage 2: final aggregation
    .write.mode("overwrite").parquet("/data/sales_by_region"))   # the action: one job

Why it matters In the Spark UI, Jobs show which actions were slow, Stages show where time and shuffle data go, and task metrics within a stage reveal skew (one task far slower than the rest) or too few tasks to use all cores.

Find the NULLs first, then decide per column whether to fill them with a sensible default, drop the rows, or keep them and handle them in logic, using fillna and dropna (also available as df.na.fill and df.na.drop), coalesce, and null-aware comparisons.

Count NULLs per column

from pyspark.sql import functions as F

df.select([F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns]).show()

Fill with defaults

df = df.fillna({"price": 0.0, "city": "unknown", "quantity": 0})

A dictionary sets a value per column; a single value only applies to columns of a matching type.

Fill with a computed value

mean_price = df.select(F.avg("price")).first()[0]
df = df.fillna({"price": mean_price})

# or per group, with a window
from pyspark.sql import Window
w = Window.partitionBy("category")
df = df.withColumn("price", F.coalesce("price", F.avg("price").over(w)))

For many numeric columns, pyspark.ml.feature.Imputer fills with the mean, median or mode.

Drop rows

df.dropna(subset=["customer_id", "order_date"])   # rows missing required fields
df.dropna(how="all")                               # rows where everything is NULL
df.dropna(thresh=3)                                # keep rows with at least 3 non-NULL values

Choose the first non-NULL value

df = df.withColumn("contact", F.coalesce("mobile", "phone", F.lit("none")))

Compare safely F.col("x") == None never matches anything, because comparing with NULL gives NULL. Use isNull(), isNotNull(), or null-safe equality eqNullSafe() (<=> in SQL), which treats two NULLs as equal; useful when joining on columns that may be NULL.

Choosing a strategy

• Required identifiers missing: drop the row, or send it to a quarantine table to investigate.
• Numeric measures: 0 only if missing really means zero; otherwise leave NULL, since aggregates such as avg ignore NULLs.
• Categories: an explicit "unknown" value keeps the rows visible in reports.
• Note which values were imputed (a flag column), so analysis can tell real from filled values.

Both group RDD records by key, but reduceByKey combines values on each partition before the shuffle, sending one partial result per key per partition across the network, while groupByKey shuffles every value and builds the full list of values per key in memory.

pairs = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])

pairs.reduceByKey(lambda x, y: x + y).collect()     # [('a', 2), ('b', 1)]
pairs.groupByKey().mapValues(sum).collect()         # same result, far more data shuffled

When to use which

• Use reduceByKey (or aggregateByKey when the result type differs from the values, such as computing sum and count for an average) for aggregations.
• Use groupByKey only when you really need all values together, such as sorting each key's events, and the groups are small.

In DataFrames This choice is made for you: df.groupBy("key").agg(F.sum("value")) plans a partial aggregation before the shuffle automatically. The equivalent trap in DataFrames is collect_list, which gathers every value per key like groupByKey. New code should use DataFrames; RDD APIs aren't available in Spark Connect or on serverless platforms.

cache() stores a DataFrame or RDD with the default storage level; persist() does the same but lets you choose the level (memory, disk, replication). Both are lazy: data is stored when the next action computes it, and kept until unpersist() or until Spark evicts it.

Default levels

• DataFrame cache(): MEMORY_AND_DISK. Partitions that don't fit in memory go to local disk.
• RDD cache(): MEMORY_ONLY. Partitions that don't fit are recomputed when needed.

Choosing a level with persist()

from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)   # usual choice for DataFrames
df.persist(StorageLevel.DISK_ONLY)         # big data reused a few times, little memory
df.persist(StorageLevel.MEMORY_ONLY)       # small, hot data; recompute if evicted
df.persist(StorageLevel.MEMORY_AND_DISK_2) # replicated on two executors, for faster recovery

PySpark's StorageLevel has no separate _SER variants: Python data is always stored serialised, so the Scala distinction doesn't apply.

SQL equivalent

CACHE TABLE recent_orders AS SELECT * FROM orders WHERE order_date >= '2026-10-01';
UNCACHE TABLE recent_orders;

When caching helps

• A DataFrame used by several actions or several branches of a job, such as a cleaned table that feeds three aggregations.
• Iterative algorithms (machine learning training loops) that re-read the same data.
• Interactive exploration of the same subset.

When it doesn't

• Data used once: caching adds work and memory use.
• Data read straight from fast columnar storage: re-reading may be as fast as the cache, especially with disk caching on platforms such as Databricks.
• When it pushes execution memory into spills or out-of-memory errors.

Good practice

• Cache after expensive steps and filters, not before.
• Trigger it deliberately if needed (df.count()), and unpersist() when finished.
• Check the Storage tab of the Spark UI for how much is cached and what fraction fits in memory.
• checkpoint() is different: it writes to reliable storage and cuts the lineage, for very long plans.
• Databricks serverless compute doesn't support the DataFrame and SQL cache APIs; write an intermediate table instead.

Actions are the operations that make Spark execute the plan and return a result to the driver or write it out. The ones I use most are count(), show(), take() and first() for inspection, collect() and toPandas() for small results, and writes such as saveAsTable() for real output, plus foreachBatch in streaming.

How I've used them

from pyspark.sql import functions as F

# Guard: skip the run when there's nothing new
new_rows = spark.read.table("bronze.events").filter(F.col("ingest_date") == run_date)
if new_rows.isEmpty():
    print("No new data")
else:
    # Data quality check before publishing
    bad = new_rows.filter(F.col("event_id").isNull()).count()
    if bad > 0:
        raise ValueError(f"{bad} events without an id")

    # Read a scalar for the next step
    max_ts = new_rows.agg(F.max("event_ts")).first()[0]

    # Publish
    new_rows.write.mode("append").saveAsTable("silver.events")

Lessons

• Each action runs the plan from the start; when several actions reuse a DataFrame, cache it or write an intermediate table.
• Avoid actions inside loops over rows; express the logic as one transformation.
• In streaming, foreachBatch lets you apply batch actions, such as a Delta MERGE, to each micro-batch.
• Keep collect() for genuinely small results; for everything else, write to storage.
• In notebooks, every display() or show() is an action too, so a cell that looks cheap can re-read a large table; check the Jobs tab when a notebook feels slow.

Call toDF() on the RDD, or pass it to spark.createDataFrame() with a schema. Giving an explicit schema is the most reliable, because inference from Python objects can choose wider types than you want.

rdd = spark.sparkContext.parallelize([("alice", 34), ("bob", 29)])

# 1. toDF with column names: types are inferred (age becomes bigint)
df1 = rdd.toDF(["name", "age"])

# 2. createDataFrame with an explicit schema
df2 = spark.createDataFrame(rdd, "name string, age int")

# 3. From Row objects, which carry their field names
from pyspark.sql import Row
df3 = spark.sparkContext.parallelize([Row(name="alice", age=34)]).toDF()

A schema can also be built with StructType and StructField when you need nullability or nested types:

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

schema = StructType([
    StructField("name", StringType(), nullable=False),
    StructField("age", IntegerType(), nullable=True),
])
df4 = spark.createDataFrame(rdd, schema)

Notes

• Inference samples rows; mixed types in a column cause errors or unexpected types, which an explicit schema avoids.
• Converting is cheap to write but moves data from Python objects into Spark's internal format.
• In Scala, import spark.implicits._ and rdd.toDF() work on RDDs of case classes and tuples.
• The RDD API isn't available in Spark Connect, so new code usually starts with DataFrames from files or tables, or builds them from Python lists with createDataFrame directly.

Read it without a schema in one of two ways. With only header, every column is a string. Add inferSchema and Spark reads the data an extra time to guess each column's type.

# Every column as a string: one pass, nothing guessed
df = spark.read.option("header", True).csv("/data/orders.csv")

# Types inferred from the data: an extra pass over the file
df = (spark.read
        .option("header", True)
        .option("inferSchema", True)
        .csv("/data/orders.csv"))
df.printSchema()

Useful options

• samplingRatio: infer from a fraction of rows to make the extra pass cheaper, at the risk of guessing wrong.
• sep, quote, escape, multiLine: for non-standard files.
• dateFormat, timestampFormat: for dates in non-ISO formats.
• enforceSchema and mode: how mismatched and malformed rows are handled.

Turning strings into types yourself A common middle ground is to read everything as strings, then cast deliberately:

from pyspark.sql import functions as F

typed = df.select(
    F.col("order_id").cast("int"),
    F.to_date("order_date", "yyyy-MM-dd").alias("order_date"),
    F.col("amount").cast("decimal(10,2)"),
    F.col("zip"),                        # identifiers with leading zeros stay strings
)

With ANSI mode, the default in Spark 4, a value that can't be cast raises an error; use try_cast (F.col("amount").try_cast("int") or try_cast(amount AS INT) in SQL) to get NULL instead.

For production pipelines, define the schema explicitly; inference is best kept for exploration.

A narrow transformation computes each output partition from a single input partition, so it needs no data movement between executors. A wide transformation needs data from many input partitions for each output partition, so Spark shuffles data across the network, and that marks the boundary between stages.

from pyspark.sql import functions as F

result = (orders
    .filter(F.col("amount") > 0)                # narrow
    .withColumn("year", F.year("order_date"))   # narrow   } one stage
    .groupBy("year").agg(F.sum("amount")))      # wide: shuffle, new stage

Nuances

• coalesce(n) is narrow because it merges existing partitions without a shuffle; repartition(n) is wide.
• A join can be narrow when one side is broadcast, or when both sides are already partitioned the same way.
• Spark reduces the cost of some wide operations automatically: aggregations combine values before the shuffle, and adaptive query execution merges small shuffle partitions.

Why it matters Shuffles are usually where Spark jobs spend their time. Filtering and aggregating early, broadcasting small tables and avoiding unnecessary repartition or orderBy calls reduce how much data is shuffled.

A temporary view is visible only in the Spark session that created it and disappears when that session ends. A global temporary view is visible to every session in the same Spark application, lives in the system database global_temp, and disappears when the application stops.

df.createOrReplaceTempView("orders")
spark.sql("SELECT COUNT(*) FROM orders")

df.createOrReplaceGlobalTempView("orders_shared")
spark.sql("SELECT COUNT(*) FROM global_temp.orders_shared")
spark.newSession().sql("SELECT COUNT(*) FROM global_temp.orders_shared")   # works
spark.newSession().sql("SELECT COUNT(*) FROM orders")                      # fails: not visible

What both have in common

• Neither stores data. A view is a saved query plan that runs again each time it is queried; cache it if you need to reuse the result.
• Neither is in the metastore or catalog, so other applications and clusters can't see them.

When to use which

• Temporary views for using a DataFrame in SQL within one notebook or job: by far the common case.
• Global temporary views for sharing between sessions of one long-running application, such as several notebooks attached to the same classic cluster.
• To share across clusters, users or restarts, create a table or a permanent view in the catalog instead.

Databricks serverless compute doesn't support global temporary views; it recommends session temporary views or tables.

Use withColumn for one new column, withColumns for several at once, or select with expressions; build the values from column expressions and built-in functions, with when/otherwise for conditions.

from pyspark.sql import functions as F

orders = (orders
    .withColumn("total", F.col("quantity") * F.col("unit_price"))
    .withColumns({
        "order_year": F.year("order_date"),
        "is_eu": F.col("country").isin("DE", "FR", "NL"),
        "band": F.when(F.col("unit_price") > 100, "high")
                 .when(F.col("unit_price") > 20, "mid")
                 .otherwise("low"),
    }))

# select keeps only the listed columns; "*" keeps the rest
orders = orders.select("*", (F.col("total") * 0.19).alias("vat"))

# SQL expressions in strings
orders = orders.selectExpr("*", "round(total * 1.19, 2) AS gross")

Window-based columns

from pyspark.sql import Window

w = Window.partitionBy("customer_id").orderBy("order_date")
orders = orders.withColumn("running_total", F.sum("total").over(w))

Good practice

• Prefer built-in functions to Python UDFs: they run in the JVM and can be optimised.
• Add many columns with one withColumns or select, not long chains of withColumn in a loop, which makes plans large and slow to analyse.
• Use F.lit() for constants and cast explicitly when types matter.
• withColumn with an existing name replaces that column.
• DataFrames are immutable: each call returns a new DataFrame, so assign the result (orders = orders.withColumn(...)), or the new column is lost.
• Name columns in snake_case without spaces or dots; names with dots need backticks in expressions, which is easy to get wrong.

Which method to use withColumn reads well for one column. withColumns takes a dictionary and adds them all in one step, which keeps the plan small. select is best when you also want to drop, rename or reorder columns at the same time, and selectExpr when the logic is easier to write in SQL. All of them produce the same optimised plan for the same expressions.

They are opposites. explode() turns each element of an array (or each key-value pair of a map) into its own row, so one row becomes many. collect_list() is an aggregate that gathers values from many rows into one array per group.

from pyspark.sql import functions as F

orders = spark.createDataFrame(
    [(1, ["apple", "pear"]), (2, []), (3, None)],
    "order_id int, items array<string>")

lines = orders.select("order_id", F.explode("items").alias("item"))
# (1, apple), (1, pear): orders 2 and 3 disappear

orders.select("order_id", F.explode_outer("items").alias("item"))
# keeps 2 and 3 with item = NULL

back = lines.groupBy("order_id").agg(F.collect_list("item").alias("items"))
# (1, [apple, pear])

Related functions

• posexplode adds each element's position.
• collect_set removes duplicates.
• inline explodes an array of structs into columns.
• array_sort, transform, filter and aggregate work on arrays without exploding, which is often cheaper.

Performance notes

• explode can multiply row counts dramatically; filter before exploding where possible.
• collect_list holds a whole group's values in memory; a key with millions of values can cause out-of-memory errors, and the order of elements is not guaranteed unless you sort, for example with array_sort or a window ordered by a column.

ORDER BY produces a total order of the whole result. SORT BY only sorts rows within each partition, so the overall output may be out of order. CLUSTER BY col is shorthand for DISTRIBUTE BY col SORT BY col: it repartitions rows so equal values go to the same partition, then sorts within each partition.

-- 1. Total order: one global sort
SELECT * FROM sales ORDER BY amount DESC;

-- 2. Sorted within each partition only
SELECT * FROM sales SORT BY amount DESC;

-- 3. Same keys together, sorted within partitions
SELECT * FROM sales CLUSTER BY customer_id;
-- same as
SELECT * FROM sales DISTRIBUTE BY customer_id SORT BY customer_id;

How ORDER BY works Spark samples the data to choose range boundaries, shuffles rows into ordered ranges, and sorts each range, so partition 1 holds the smallest values. It is the most expensive of the three; with LIMIT, Spark uses a cheaper top-N sort.

When to use each

• ORDER BY for final results that people read or that must be ordered, preferably after aggregation, when data is small.
• SORT BY when only per-partition order matters, such as preparing sorted output files where each file is sorted.
• DISTRIBUTE BY or CLUSTER BY to group all rows with the same key in one partition before per-key processing, or so each written file contains whole keys, which improves compression and data skipping.

Pitfalls

• Don't rely on SORT BY or CLUSTER BY for an ordered report: the output is only ordered within partitions.
• Ordering isn't preserved by later shuffles; sort as the last step.
• CLUSTER BY in a SELECT is unrelated to CLUSTER BY in CREATE TABLE, which (for Delta and some other table formats) defines liquid clustering, a storage layout.

Parquet is a columnar, compressed, self-describing file format, and Spark is built to exploit it. Queries read only the columns they need, skip data using the statistics stored in each file, decompress quickly, and get the schema from the file itself. That usually makes Parquet many times faster and smaller than CSV or JSON for analytics.

1. Columnar storage: read only what you need Values of each column are stored together. SELECT customer_id, amount on a 100-column table reads two columns' data, not the whole rows. Spark's column pruning does this automatically.

2. Predicate pushdown and data skipping Each file is split into row groups, and each row group stores the minimum and maximum (and null counts) for every column. Spark pushes filters such as amount > 1000 into the reader (spark.sql.parquet.filterPushdown, on by default) and skips row groups that can't match. Sorting or clustering data by commonly filtered columns makes this much more effective.

3. Compression and encoding Similar values stored together compress well. Parquet uses dictionary and run-length encoding plus a codec: Spark writes Snappy by default, and zstd usually gives smaller files at similar speed. Files are commonly several times smaller than the same data in CSV.

4. Schema stored with the data Column names and types are in the file footer, so there is no schema inference and no type guessing; nested structures (structs, arrays, maps) are supported natively.

5. Fast reading in Spark Spark's vectorised Parquet reader decodes batches of column values at a time, which works well with whole-stage code generation.

6. Splittable Files can be read in parallel by row group, even when compressed.

7. Partition discovery Directory layouts such as /sales/year=2026/month=10/ are read as partition columns, and filters on them skip whole directories.

8. The foundation of table formats Delta Lake, Apache Iceberg and Apache Hudi usually store data as Parquet files and add transactions, schema evolution, time travel and file-level statistics on top. In practice you usually use one of them rather than plain Parquet folders, but the speed comes from Parquet underneath.

9. Interoperability Almost every data tool reads Parquet: pandas, Polars, DuckDB, Trino, BigQuery, Snowflake and the cloud warehouses.

Limitations

• Files are immutable: updating a row means rewriting a file, which is why table formats exist.
• Writing is slower than appending to a row format, so it's less suited to many tiny writes.
• Many small files ruin performance; aim for files of roughly 128 MB to 1 GB.
• Schema changes across files need mergeSchema or a table format to manage.

df.write.mode("overwrite").option("compression", "zstd").partitionBy("order_date").parquet("/data/orders")
spark.read.parquet("/data/orders").filter("order_date = '2026-10-01' AND amount > 1000").select("order_id", "amount")

Use DataFrame.drop() with one or more column names. It returns a new DataFrame without those columns and ignores names that don't exist.

df2 = df.drop("note")
df2 = df.drop("note", "internal_id", "tmp")   # several at once
df2 = df.drop(df["note"])                      # by column reference, useful after joins

cols_to_drop = [c for c in df.columns if c.startswith("tmp_")]
df2 = df.drop(*cols_to_drop)

Alternatives

• Keep a list instead of removing one: df.select("order_id", "amount"). For wide tables where you want most columns, select with a comprehension works: df.select([c for c in df.columns if c != "note"]).
• In SQL: SELECT * EXCEPT (note) FROM orders (supported by Spark SQL and Databricks).

After a join with duplicate names When both sides have a column called id, drop the one you don't want by reference: joined.drop(right["id"]). Joining with on="id" (a column name rather than an expression) keeps only one id column in the first place.

Dropping a column from a table df.drop only affects the DataFrame. To remove a column from a stored table, use the table's DDL; in Delta Lake, ALTER TABLE orders DROP COLUMN note requires column mapping to be enabled on the table, otherwise you rewrite the table without it.

Remember to assign the result: df.drop("note") on its own does nothing to df.

The DAG (directed acyclic graph) is the plan Spark builds of how each result depends on its inputs. It lets Spark see the whole computation before running it, group operations that don't need to move data into stages, run independent work in parallel, and recover lost data by recomputing only what depends on it.

How execution flows through the DAG

1. Building: transformations add nodes to a logical plan; nothing runs. For DataFrames and SQL, the Catalyst optimizer rewrites this plan and chooses a physical plan.
2. Action: an action (count, write, collect) submits a job to the DAG scheduler.
3. Stage creation: the DAG scheduler walks back from the final result. Operations connected by narrow dependencies (filter, select, map) are pipelined into one stage; each shuffle dependency (groupBy, shuffle joins, repartition) creates a stage boundary.
4. Ordering: stages are submitted when their parent stages have finished; independent branches, such as scanning two tables for a join, run at the same time.
5. Tasks: each stage becomes one task per partition, sent to executors by the task scheduler.
6. Recovery: if a task fails it is retried; if shuffle output is lost, the scheduler re-runs only the parent stage that produced it.

What the DAG enables

• Optimisation of the whole job: filters pushed to the source, projections pruned, operations fused into generated code.
• Pipelining: narrow operations run together on each partition with no intermediate data written.
• Minimal shuffles: the shuffle boundaries are explicit, which is where data moves.
• Fault tolerance: lineage edges say exactly what to recompute.
• Runtime re-planning: adaptive query execution changes the remaining stages once shuffle sizes are known.

Example

from pyspark.sql import functions as F

orders = spark.read.parquet("/data/orders").filter(F.col("amount") > 0)
customers = spark.read.parquet("/data/customers")
report = (orders.join(customers, "customer_id")        # shuffle (or broadcast)
                .groupBy("country")                    # shuffle
                .agg(F.sum("amount").alias("revenue")))
report.write.mode("overwrite").parquet("/out/revenue_by_country")

The scans of orders and customers are separate stages that run in parallel; the join and the aggregation add stage boundaries, unless the customers table is broadcast, in which case the join happens inside the orders stage.

Seeing it The Spark UI shows a DAG visualisation for every job and stage, and the SQL tab shows the physical plan with metrics per operator. Reading it is the quickest way to see where shuffles happen and which stage is slow.

repartition(n) performs a full shuffle to produce exactly n roughly equal partitions, and can increase or decrease the count or partition by columns. coalesce(n) only decreases the number of partitions by merging existing ones without a shuffle, so it is cheaper but can leave partitions uneven.

df = spark.range(0, 1_000_000, numPartitions=8)

df.repartition(16)              # shuffle → 16 even partitions
df.repartition("country")       # shuffle by column (hash partitioning), spark.sql.shuffle.partitions partitions
df.repartition(32, "country")   # by column, 32 partitions
df.coalesce(2)                  # merge 8 into 2, no shuffle
df.coalesce(16)                 # still 8: coalesce can't increase

When to use repartition

• Too few partitions for the cluster (for example after reading one large gzip file) or very uneven partitions.
• Before heavy processing by key, or before a partitioned write, to group rows by the partition column so each task writes few files: df.repartition("event_date").write.partitionBy("event_date").
• repartitionByRange when you need range-ordered partitions.

When to use coalesce

• Reducing the number of output files after a filter has shrunk the data: df.coalesce(10).write....
• Any decrease where small imbalance doesn't matter.

The coalesce trap coalesce(1) before writing one file makes the whole last stage, sometimes including the expensive work before it, run in a single task. For large data, repartition(1) is often faster, because the upstream work stays parallel and only the final result is shuffled to one task. Better still, avoid single files for big data.

In SQL and with adaptive query execution Hints do the same in SQL: /*+ REPARTITION(16) */, /*+ COALESCE(4) */, and /*+ REBALANCE */, which lets adaptive query execution produce evenly sized output partitions. With AQE on (the default), Spark already merges small shuffle partitions, so manual repartitioning is needed less often than it used to be.

Adaptive query execution (AQE) is Spark SQL's runtime re-optimisation: it uses the actual statistics of data produced during a query to adjust the rest of the query plan. It solves the problem that plans made before execution rely on estimates that are often wrong, which leads to badly sized partitions, the wrong join strategy and skewed tasks. AQE has been on by default since Spark 3.2 (spark.sql.adaptive.enabled).

The problem it solves Before AQE, Spark fixed the whole plan up front:

• Partition count: one value of spark.sql.shuffle.partitions (200 by default) for every shuffle in every query. Small data got thousands of tiny tasks; large data got huge partitions that spilled to disk.
• Join strategy: chosen from estimates. A table that would shrink to 2 MB after a filter might still be joined with an expensive sort-merge join, because the planner thought it was large.
• Skew: one key with most of the rows created a single task that ran for hours while the others finished in minutes.
• Missing statistics: many tables have none, so estimates were guesses.

How it works The query is divided into stages at exchanges (shuffles and broadcasts). Spark runs the stages whose inputs are ready, collects the size of each output partition, then re-optimises and re-plans the remaining part of the query before continuing.

What it does at runtime

1. Coalesces post-shuffle partitions into fewer, larger ones (spark.sql.adaptive.coalescePartitions.enabled, target size 64 MB by default).
2. Converts sort-merge joins to broadcast hash joins when one side is small enough at runtime, reading shuffle output locally.
3. Converts sort-merge joins to shuffled hash joins when all partitions are small enough (off by default; enabled with spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold).
4. Splits skewed partitions in sort-merge joins: a partition is skewed if it is larger than 5 times the median and larger than 256 MB.
5. Splits skewed partitions when rebalancing output (with the REBALANCE hint).

Effect in practice

• Less manual tuning: set a generous initial partition number and let AQE coalesce.
• Faster joins when filters make one side small.
• Fewer straggler tasks from skew.

Example settings

spark.conf.set("spark.sql.adaptive.enabled", "true")                          # default
spark.conf.set("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "2000")
spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")  # respect the 64 MB target on busy clusters

By default (parallelismFirst true), coalescing favours parallelism and only respects a 1 MB minimum partition size; the Spark documentation recommends setting it to false on a busy cluster so partitions reach the advisory size.

What it doesn't do It doesn't change how data is read in the first stage, doesn't apply to streaming queries, and doesn't fix skew in aggregations or in joins that aren't sort-merge joins. See the question on AQE's limitations.

Accumulators and broadcast variables are Spark's two kinds of shared variables. An accumulator lets tasks add to a value that only the driver can read, such as a count of bad records. A broadcast variable sends a read-only value, such as a lookup table, to each executor once instead of with every task.

Accumulators Tasks can only add to them; the driver reads the result after an action.

bad_rows = spark.sparkContext.accumulator(0)

def check(row):
    if row["amount"] is None:
        bad_rows.add(1)

orders.foreach(check)          # an action
print(bad_rows.value)          # read on the driver

• Built-in numeric accumulators exist; Scala and Java also have collection accumulators, and you can define custom ones (AccumulatorParam in Python, AccumulatorV2 in Scala).
• Reliability: updates made inside actions are applied exactly once per task. Updates made inside transformations (map, filter, a UDF) can be applied more than once if a task or stage is re-executed, or not at all if the transformation is never computed, so use them there only for rough debugging counts.
• They are lazy: nothing is counted until an action runs.
• For data quality counts in DataFrame code, a separate aggregation (df.filter(...).count()) or metrics written to a table are usually clearer.

Broadcast variables A read-only value cached on each executor.

rates = {"EUR": 1.0, "USD": 0.92, "GBP": 1.17}
b_rates = spark.sparkContext.broadcast(rates)

from pyspark.sql import functions as F, types as T
@F.udf(T.DoubleType())
def to_eur(amount, ccy):
    return amount * b_rates.value.get(ccy, 0.0) if amount is not None else None

orders.withColumn("amount_eur", to_eur("amount", "currency"))
b_rates.unpersist()            # when no longer needed

• Spark sends the value efficiently (in pieces, peer to peer), once per executor, instead of serialising it into every task's closure.
• Values must fit in executor and driver memory and must not be modified.
• unpersist() removes copies from executors; destroy() releases them everywhere.

The DataFrame equivalent: broadcast joins For lookup tables, the idiomatic DataFrame approach is a broadcast join rather than a broadcast variable in a UDF:

rates_df = spark.createDataFrame(list(rates.items()), ["currency", "rate"])
orders.join(F.broadcast(rates_df), "currency", "left")

This keeps the work inside Spark's optimised engine. Spark broadcasts small tables automatically (below 10 MB by default).

Environment note Both are part of the classic SparkContext API. Spark Connect and serverless environments that use it (such as Databricks serverless compute) don't support sparkContext accumulators and broadcast variables; use aggregations and broadcast joins there.

Set the sep option (also accepted as delimiter) when reading the CSV file.

df = (spark.read
        .option("header", True)
        .option("sep", "#")
        .schema("id INT, name STRING, amount DOUBLE")
        .csv("/data/orders_hash.csv"))

or with format:

df = spark.read.format("csv").options(header=True, sep="#").load("/data/orders_hash.csv")

Options that usually go with it

• header: use the first line as column names.
• schema(...): give the types explicitly. inferSchema=True works but reads the data an extra time and can guess wrong (for example, turning a zip code 01234 into the number 1234).
• quote and escape: if fields can contain #, they must be quoted (by default with "), or the file must escape them.
• encoding: for files that aren't UTF-8.
• mode: PERMISSIVE (default; malformed values become null), DROPMALFORMED or FAILFAST. With PERMISSIVE, add columnNameOfCorruptRecord and a string column with that name to the schema to keep the raw text of bad rows.
• multiLine: for quoted fields that contain line breaks.
• nullValue, dateFormat and timestampFormat for how empty values and dates are written.

Multi-character and special delimiters sep accepts more than one character, such as "||" or "#|", and escape sequences such as tab ("\t") and Unicode characters such as "\u0001".

In SQL

CREATE TEMPORARY VIEW orders_hash
USING csv
OPTIONS (path '/data/orders_hash.csv', header 'true', sep '#');

Writing with a custom delimiter

df.write.option("header", True).option("sep", "#").csv("/out/orders_hash")

When the delimiter isn't consistent If lines mix delimiters or contain free text, read them with spark.read.text and parse with split or regexp_extract instead.

Select nested struct fields with dot notation, turn arrays into rows with explode (or explode_outer to keep rows with empty arrays), and repeat until every column is a simple type. For wide or changing schemas, write a function that walks the schema and flattens structs automatically.

Example input

{"order_id": 1, "customer": {"id": 7, "address": {"city": "Berlin"}},
 "items": [{"sku": "A", "qty": 2}, {"sku": "B", "qty": 1}]}

Explicit flattening

from pyspark.sql import functions as F

orders = spark.read.json("/data/orders.json")      # multiLine=True for one JSON document across many lines

flat = (orders
    .select(
        "order_id",
        F.col("customer.id").alias("customer_id"),
        F.col("customer.address.city").alias("city"),
        F.explode_outer("items").alias("item"))
    .select("*", "item.*")
    .drop("item"))

Result: one row per item, with columns order_id, customer_id, city, qty, sku. An order with an empty items array keeps one row with nulls, because of explode_outer.

Flattening structs automatically

from pyspark.sql import types as T

def flatten_structs(df):
    while True:
        structs = [f.name for f in df.schema.fields if isinstance(f.dataType, T.StructType)]
        if not structs:
            return df
        cols = []
        for f in df.schema.fields:
            if f.name in structs:
                cols += [F.col(f"`{f.name}`.`{s.name}`").alias(f"{f.name}_{s.name}")
                         for s in f.dataType.fields]
            else:
                cols.append(F.col(f"`{f.name}`"))
        df = df.select(cols)

On the example this gives customer_address_city, customer_id, items and order_id. Arrays are left alone on purpose: decide per array whether to explode it.

Decisions to make

• Exploding multiplies rows: exploding two independent arrays in the same table creates a cross product. Usually each array becomes its own table (orders, order_items), joined by the parent key, which is how it would be modelled anyway.
• Keep the parent key on every child row.
• Positions: posexplode keeps the array index if order matters.
• Maps: explode on a map gives key and value columns.
• Schema: give an explicit schema (or parse a string column with from_json and a schema) so new or missing fields don't change the output unexpectedly.

Alternatives to full flattening Lakehouse tables handle nested types well, so you can keep structs and select fields when needed. On Databricks, the VARIANT type (also in open-source Spark 4) stores semi-structured JSON efficiently and is queried with path expressions, which suits data whose shape changes often.

Transformations describe a new dataset derived from an existing one (select, filter, join, groupBy) and are lazy: Spark only records them in a plan. Actions (count, collect, show, write) ask for a result, which makes Spark optimise the plan and run a job on the cluster.

Transformations

• Return a new DataFrame or RDD; the original is unchanged (data is immutable).
• Lazy: nothing is read or computed yet.
• Two kinds:
  • Narrow: each output partition depends on one input partition (select, filter, withColumn, map, union). They run together in one stage without moving data.
  • Wide: output partitions depend on many input partitions (groupBy().agg(), join without broadcast, distinct, orderBy, repartition). They need a shuffle and start a new stage.

Actions

• Return a value to the driver (count, collect, first, take, show, toPandas) or write data out (write.save, write.saveAsTable, foreach).
• Trigger a job: the scheduler turns the plan into stages and tasks.
• Each action runs the whole lineage again unless intermediate results are cached.

from pyspark.sql import functions as F

orders = spark.read.parquet("/data/orders")                 # nothing read yet
big = orders.filter(F.col("amount") > 1000)                 # transformation (narrow)
by_country = big.groupBy("country").count()                 # transformation (wide)
by_country.show()                                           # action: job runs
by_country.write.mode("overwrite").parquet("/out/big_by_country")   # action: runs again

The second action recomputes everything from the source. If both are needed, by_country.cache() before the first action avoids that, at the cost of memory.

Why laziness helps

• Spark sees the whole pipeline before running it, so it can push the filter into the Parquet scan, read only the needed columns and fuse operations together.
• No intermediate results are materialised unless needed.
• Lineage lets lost partitions be recomputed after failures.

Common surprises

• Errors in a transformation (a bad cast, a missing file) appear only when an action runs, in a different line from where they were written. Since Spark 4, ANSI mode makes invalid casts raise errors, which makes this more visible.
• collect() and toPandas() bring all rows to the driver and can exhaust its memory; use limit, take or write the result instead.
• Calling count() just to "check" data runs a full job.
• Some operations behave like actions behind the scenes: reading CSV with inferSchema, or checkpoint(), run jobs immediately.

Quick classification

Stop moving the data to the driver. The driver is one machine with 1 GiB of heap by default, and spark.driver.maxResultSize (also 1 GiB) aborts any action whose results would exceed it. Keep the work on the executors: write the result to storage, aggregate it down before collecting, or take a sample with limit(). Raising the 2 settings works as a stopgap, but Spark's own documentation warns that a high limit can crash the driver instead.

1. Choose by what you need the data for

2. Before and after

# Before: millions of rows into one process
rows = df.filter("event_date = '2026-10-21'").collect()
for r in rows:
    send(r)

# After: the executors write; another job or service reads the files
(df.filter("event_date = '2026-10-21'")
   .write.mode("overwrite")
   .parquet("s3://bucket/exports/2026-10-21/"))

3. The error text tells you which limit

• Total size of serialized results of N tasks (X MB) is bigger than spark.driver.maxResultSize: the guard stopped it before the driver crashed.
• OutOfMemoryError in the driver log: the data, or a broadcast built on the driver, did not fit in its heap.

4. If you must raise the limits

Set both together and with room: spark.driver.memory well above spark.driver.maxResultSize, since the driver must hold the deserialized data too. Then plan the code change; a dataset that grows will cross the new limit as well.

What interviewers want: the driver as a single machine with small defaults, the 2 settings and what each error means, writing or aggregating instead of collecting, limit and toLocalIterator for inspection, and raising limits only as a stopgap.

Read the source table with spark.read.table() (or format("delta").load() for a path), chain DataFrame transformations, and write the result with format("delta") and saveAsTable(). For incremental loads, use a MERGE instead of overwriting the whole target.

from pyspark.sql import functions as F

# 1. Read: by table name (Unity Catalog or the metastore) rather than a storage path
orders = spark.read.table("bronze.orders")

# 2. Transform
silver = (
    orders
    .filter(F.col("status") == "active")
    .dropDuplicates(["order_id"])
    .withColumn("order_ts", F.to_timestamp("order_ts"))
    .withColumn("order_date", F.to_date("order_ts"))
    .withColumn("amount", F.col("amount").cast("decimal(12,2)"))
    .withColumnRenamed("email", "customer_email")
)

# 3. Write a full refresh to another Delta table
(silver.write
    .format("delta")
    .mode("overwrite")
    .partitionBy("order_date")      # only for large tables with a low-cardinality column
    .saveAsTable("silver.orders"))

Incremental upsert instead of overwrite

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "silver.orders")
(target.alias("t")
    .merge(silver.alias("s"), "t.order_id = s.order_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Points worth mentioning

• Delta checks the schema on write: a DataFrame with different columns fails unless you allow evolution with .option("mergeSchema", "true").
• mode("overwrite") replaces the data atomically; readers see the old or the new version, never a mix, and time travel can recover the old version.
• For a path-based table, use spark.read.format("delta").load(path) and .save(path).
• In Databricks this pattern is the bronze-to-silver step of the medallion architecture; for pipelines that run continuously, Lakeflow declarative pipelines or Structured Streaming with readStream on the Delta source process only new rows.

Spark recovers from failures mainly by recomputing lost data from its lineage, the recorded sequence of transformations that produced it, and by retrying failed tasks, rather than by copying all data as it goes.

1. Lineage and recomputation Every DataFrame or RDD knows the steps that built it from its source. If an executor dies and its partitions are lost, the driver reschedules the tasks that produce those partitions, re-reading the input and re-running the transformations for those partitions only.

2. Task retries A failed task is retried on another executor; by default a job fails only after the same task fails 4 times (spark.task.maxFailures). Optional speculative execution starts duplicate copies of very slow tasks.

3. Shuffle files Map-stage outputs are written to local disk, so a failure in a later stage doesn't redo earlier stages if their shuffle files survive. If they're lost with an executor, Spark reruns the stage that produced them. An external shuffle service or shuffle tracking lets executors be removed without losing them.

4. Checkpointing and replication For very long lineages (iterative algorithms, long streaming jobs), checkpoint() saves data to reliable storage and cuts the lineage, so recovery doesn't replay everything. Storage levels ending in _2 keep cached partitions on two executors.

5. Reliable storage underneath Input and output live in replicated storage such as HDFS or cloud object storage, and table formats such as Delta Lake make writes atomic, so a failed job doesn't leave half-written tables.

6. Streaming Structured Streaming records progress and state in a checkpoint location; with a replayable source and an idempotent or transactional sink, a restarted query continues with exactly-once results.

What isn't covered If the driver fails, the application fails, unless the cluster manager restarts it (cluster deploy mode with supervision) or, for streaming, a new run resumes from the checkpoint.

The driver is the process that runs your program's main code and plans and coordinates the work; executors are processes on worker nodes that run the tasks and hold data. The driver decides what to do; executors do it.

Driver

• Runs the application code and creates the SparkSession.
• Turns DataFrame operations and SQL into logical and physical plans (with the Catalyst optimizer), then into jobs, stages and tasks.
• Requests executors from the cluster manager (YARN, Kubernetes, standalone, or a managed platform such as Databricks).
• Schedules tasks on executors, preferring executors near the data, and tracks their progress.
• Retries failed tasks and resubmits stages when shuffle data is lost.
• Collects results of actions such as collect() and count(), and holds broadcast variables before sending them out.
• Serves the Spark UI. A failed driver ends the application, so don't overload it: collect() on a large DataFrame is the classic way to run it out of memory.

Executors

• Run tasks: each core runs one task at a time on one partition.
• Read input, compute transformations and write output.
• Write and serve shuffle files for the next stage.
• Store cached data in memory or on disk.
• Report status and metrics to the driver.
• In PySpark, start Python worker processes for Python UDFs; that memory lives outside the JVM heap.

Example: counting words

from pyspark.sql import functions as F

counts = (spark.read.text("/data/books")
    .select(F.explode(F.split(F.lower("value"), r"\W+")).alias("word"))
    .filter(F.col("word") != "")
    .groupBy("word").count())
counts.write.mode("overwrite").parquet("/data/word_counts")

The driver builds a plan with two stages. In the first, each executor reads part of the files, splits lines into words and counts words locally. The shuffle sends each word's partial counts to one executor, which adds them up in the second stage and writes its part of the output. The driver only coordinates; no data passes through it.

Lazy evaluation means transformations such as filter, select and join don't run when you call them. Spark records them as a plan, and only when an action (count, collect, show, a write) needs a result does it optimise the whole plan and execute it.

from pyspark.sql import functions as F

df = spark.read.parquet("/data/orders")              # nothing read yet (only metadata)
active = df.filter(F.col("status") == "active")       # recorded
small = active.select("order_id", "amount")           # recorded

small.explain()        # shows the optimised plan, still no data processed
small.count()          # action: now Spark reads and computes

Why it matters

1. Whole-plan optimisation Because Spark sees all the steps before running them, the Catalyst optimizer can:

• push the filter down to the Parquet scan, so files and row groups that can't match are skipped;
• read only the two needed columns;
• combine consecutive operations into one stage, with generated code and no intermediate data written;
• pick join strategies based on table sizes.

2. No wasted work Data that no action uses is never computed. df.limit(10).show() can stop after reading a few files.

3. Fault tolerance The recorded plan is also the lineage Spark uses to recompute lost partitions.

Consequences to know

• Errors appear late: a typo in a column name is caught when the plan is analysed, but bad data (such as a value that can't be cast) fails only when an action runs, sometimes far from the line that caused it.
• Each action recomputes the plan from the start. If a DataFrame is used by several actions, cache() it, or the source is read again each time.
• Timing a transformation line measures nothing; time the action.
• Some operations do run work early: schema inference reads data when the DataFrame is defined.

Transformations describe how to build a new DataFrame or RDD from an existing one and are lazy; actions ask for a result (returned to the driver or written to storage) and make Spark execute all the transformations they depend on.

Transformations

• Return a new DataFrame or RDD; the original is unchanged, because they are immutable.
• Are recorded in a plan, not executed.
• Come in two kinds:
  • Narrow: each output partition depends on one input partition, so no data moves between executors. Examples: select, filter/where, withColumn, drop, map, flatMap, union, coalesce.
  • Wide: output partitions need data from many input partitions, so Spark shuffles data across the network and starts a new stage. Examples: groupBy().agg(), join (unless broadcast), distinct, orderBy, repartition, reduceByKey.

Actions

• Trigger a job.
• Return values to the driver: count(), collect(), first(), take(n), show(), toPandas().
• Or write output: write.save(), write.saveAsTable(), foreach(), foreachPartition().

Example

from pyspark.sql import functions as F

orders = spark.read.parquet("/data/orders")

summary = (orders
    .filter(F.col("amount") > 0)                    # narrow
    .withColumn("year", F.year("order_date"))       # narrow
    .groupBy("year", "region")                      # wide: shuffle
    .agg(F.sum("amount").alias("revenue")))

summary.show(5)                                     # action: job runs
summary.write.mode("overwrite").parquet("/out/rev") # action: runs again

Nothing executes until show(). The second action recomputes everything unless summary is cached.

Nuances

• collect() and toPandas() bring all data to the driver; use them only on small results. Prefer show(), take() or writing to storage.
• Wide transformations are expensive because of the shuffle; filter and aggregate early to shrink what is shuffled, and broadcast small tables in joins.
• cache() and persist() are lazy too: they mark data to be kept, and it is stored by the next action.
• Calling explain() or reading df.schema doesn't run a job, but schema inference on CSV or JSON does read data.
• In notebooks, display commands are actions, so each displayed cell can run a job.

How to tell them apart quickly If the method returns a DataFrame or RDD, it is a transformation; if it returns a Python value, a list of rows or nothing (because it writes somewhere), it is an action. In the Spark UI, every action appears as one or more jobs, so a notebook that shows dozens of jobs for one cell usually has more actions than intended.

Transformations define what to compute; actions decide when, which is what lets Spark optimise the whole plan before running it.

A broadcast join sends a full copy of the small table to every executor, so each executor joins its part of the large table locally. The large table is never shuffled, which removes the most expensive part of a join and also avoids skew problems on the join key.

How it works

1. The driver collects the small table and broadcasts it to all executors.
2. Each executor builds an in-memory hash table from it.
3. Each task scans its partition of the large table and looks up matches. There is one stage for the join instead of two shuffle stages plus a sort, as in a sort-merge join.

Performance benefits

• No shuffle of the large table: no network transfer, no shuffle files, no sort.
• Fewer stages: lower latency and less scheduling overhead.
• Immune to key skew: hot keys in the large table don't concentrate on one task.
• Works with dynamic partition pruning: the broadcast result can also be used to skip partitions of the large table.

When Spark uses it

• Automatically when the estimated size of one side is under spark.sql.autoBroadcastJoinThreshold (10 MB by default).
• At runtime with adaptive query execution: if a side turns out to be small after filtering, AQE switches a planned sort-merge join to a broadcast join.
• When you ask for it:

from pyspark.sql.functions import broadcast

result = sales.join(broadcast(stores), "store_id")

SELECT /*+ BROADCAST(s) */ *
FROM sales f JOIN stores s ON f.store_id = s.store_id;

Limits and risks

• The small side must fit in the driver's and every executor's memory; forcing a broadcast of a large table causes out-of-memory errors or timeouts (spark.sql.broadcastTimeout, 300 seconds by default).
• The smaller side can't be the side whose unmatched rows are kept: in a left outer join, only the right side can be broadcast.
• Size estimates for complex subqueries can be wrong; check the plan for BroadcastHashJoin.
• Raising the threshold a lot shifts load to the driver.

Practical use cases

• Fact tables joined to dimension tables: products, stores, currencies, calendars.
• Lookup and mapping tables: country codes, status codes.
• Filtering a large table by a small list of keys (often a left semi join).
• Enriching streaming data with a small static table in Structured Streaming.

Check the result with explain(): BroadcastHashJoin with BroadcastExchange on the small side confirms it.

Read all the files in one call, with a directory, a glob pattern or a list of paths, so Spark plans one scan and reads files in parallel. Combine DataFrames that were read separately with unionByName, and fix small-file and schema problems at the same time.

One read for many files

# whole directory (Spark also discovers partition folders such as year=2026/)
df = spark.read.parquet("/data/events/")

# a pattern, or a list of paths
df = spark.read.option("header", True).csv("/landing/sales/2026-*/*.csv")
df = spark.read.json(["/landing/app1/", "/landing/app2/"])

# filters and recursion
df = (spark.read
      .option("recursiveFileLookup", "true")
      .option("pathGlobFilter", "*.parquet")
      .parquet("/landing/raw/"))

Give an explicit schema for CSV and JSON so Spark doesn't read the files twice to infer one.

Keep track of the source file _metadata is a hidden column on file sources:

df = df.select("*", "_metadata.file_path", "_metadata.file_modification_time")

Combining separately read DataFrames

combined = df_2025.unionByName(df_2026, allowMissingColumns=True)

union matches columns by position, which silently mixes up columns if the order differs; unionByName matches by name, and allowMissingColumns fills missing ones with NULL.

Schema differences in Parquet .option("mergeSchema", "true") merges compatible schemas across Parquet files. It is slower, so enable it only when needed.

Efficiency

• Small files: thousands of tiny files mean slow listing and many tiny tasks. Spark packs small files into partitions up to spark.sql.files.maxPartitionBytes (128 MB), but compacting the files is the real fix.
• Partition pruning: if files are laid out in partition folders, filters on those columns skip whole folders.
• Incremental loads: Structured Streaming's file source, or Auto Loader in Databricks, processes only new files instead of re-reading all of them.
• Write the combined result to a columnar table (Parquet or Delta) so later jobs don't repeat the work.

Create a fact table partitioned by the join key and a small dimension table, join them with a filter only on the dimension, and look at the plan and the partitions read.

from pyspark.sql import functions as F

# Fact table: 30 days, partitioned by date_id
(spark.range(0, 1_000_000)
    .select((F.col("id") % 30).alias("date_id"), F.col("id").alias("amount"))
    .write.mode("overwrite").partitionBy("date_id").saveAsTable("sales"))

# Dimension table: which quarter each day belongs to
(spark.range(0, 30)
    .select(F.col("id").alias("date_id"),
            F.when(F.col("id") < 5, "Q1").otherwise("Q2").alias("quarter"))
    .write.mode("overwrite").saveAsTable("dates"))

The query

q = spark.sql("""
    SELECT d.quarter, SUM(s.amount) AS revenue
    FROM sales s
    JOIN dates d ON s.date_id = d.date_id
    WHERE d.quarter = 'Q1'
    GROUP BY d.quarter
""")
q.explain("formatted")
q.show()

What to look for

• On the sales scan: PartitionFilters: [isnotnull(date_id), dynamicpruningexpression(date_id IN dynamicpruning#..)].
• A SubqueryBroadcast dynamicpruning#.. node reusing the broadcast of the filtered dates table.
• In the Spark UI's SQL tab, the sales scan reports reading 5 of the 30 partitions.

Compare with DPP off

spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "false")
spark.sql("...same query...").explain("formatted")   # no dynamicpruningexpression; all 30 partitions read
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")

Showing when it doesn't apply

• Set spark.sql.autoBroadcastJoinThreshold to -1: the join becomes a sort-merge join, the scan shows dynamicpruningexpression(true), and all partitions are read, because there is no broadcast to reuse.
• Filter on a column of sales instead of dates: that is ordinary filtering, not DPP.

Explaining these contrasts in an interview shows you understand both how DPP works and how to verify it, rather than assuming it happened.

Data skew is an uneven distribution of data across partitions, usually because a few key values have far more rows than the rest. The tasks that process those partitions run much longer, or run out of memory, while the others finish quickly, so the whole stage waits for a few stragglers.

Where it comes from

• Joins and aggregations on keys with hot values: a few large customers, a default value such as "unknown", or NULL keys.
• Time-based data with spikes, such as a sales day.
• Input files of very different sizes, or a repartition on a skewed column.

How to spot it

• In the Spark UI's stage details, the maximum task duration and shuffle read size are far above the median.
• Most tasks finish and the stage hangs on the last few.
• Out-of-memory errors or heavy spill on only some tasks.
• To find hot keys: df.groupBy("key").count().orderBy(F.desc("count")).show().

How to address it

1. Adaptive query execution (on by default) splits skewed partitions in sort-merge and shuffled hash joins: a partition over 5 times the median and over 256 MB is split, and the matching rows on the other side are replicated.
2. Broadcast joins when one side is small: no shuffle, so no skew.
3. Handle NULL and default keys separately: filter them out before inner joins, or spread them with random values in outer joins.
4. Salting: add a random number to hot keys so their rows spread over several partitions, then combine the partial results (two-step aggregation), or replicate the matching rows of the other side for joins.
5. Isolate hot keys: process the few biggest keys in a separate path, often with a broadcast join, and union the results.
6. Better partitioning when writing: avoid partitioning tables by skewed columns; in Delta, compaction and clustering keep files evenly sized.

Check the fix Compare task duration distributions before and after. A good fix brings the maximum close to the median without adding much total work; salting, in particular, adds extra steps, so use it only where skew is real.

In Scala, call .as[T] on a DataFrame, where T is a case class whose fields match the column names and types; Spark finds the encoder through import spark.implicits._. In Java, use .as(Encoders.bean(MyClass.class)). In Python there is no typed Dataset API: a PySpark DataFrame is already the Dataset of rows.

Scala

case class Order(orderId: Long, customer: String, amount: Double)

import spark.implicits._

val df = spark.read.parquet("/data/orders")         // DataFrame = Dataset[Row]
val orders: Dataset[Order] = df
  .withColumnRenamed("order_id", "orderId")         // names must match the fields
  .as[Order]

val big = orders.filter(o => o.amount > 1000)       // compile-time checked field access

Java

Dataset<Order> orders = df.as(Encoders.bean(Order.class));

Rules and nuances

• Columns are matched to fields by name, not by position; a missing column fails at analysis time. Extra columns are kept in the underlying data until you select fields.
• Types must be compatible; a nullable column mapped to a primitive such as Double fails at runtime on a NULL, so use Option[Double] for nullable fields.
• Typed lambdas such as filter(o => o.amount > 1000) are opaque to the Catalyst optimizer: Spark must deserialise each row into an object, and it can't push the filter down to the source. Column expressions (filter($"amount" > 1000)) stay fully optimised, so many teams use Datasets for type safety at the boundaries and column expressions inside.
• Going back is ds.toDF().

Why use a Dataset Compile-time checks of field names and types in Scala and Java code, and domain objects that are easy to pass around. In Python, use explicit schemas, type hints and tests for similar safety.

A left semi join returns the rows of the left table that have at least one match in the right table, with only the left table's columns and no duplicates from multiple matches. A left anti join returns the rows of the left table that have no match. They are Spark's join forms of EXISTS and NOT EXISTS.

customers = spark.createDataFrame([(1, "a"), (2, "b"), (3, "c")], "customer_id int, name string")
orders = spark.createDataFrame([(1,), (1,), (3,)], "customer_id int")

customers.join(orders, "customer_id", "left_semi").show()   # customers 1 and 3, once each
customers.join(orders, "customer_id", "left_anti").show()   # customer 2: never ordered

SELECT * FROM customers c LEFT SEMI JOIN orders o ON c.customer_id = o.customer_id;
SELECT * FROM customers c LEFT ANTI JOIN orders o ON c.customer_id = o.customer_id;

-- equivalent forms Spark rewrites to these joins
SELECT * FROM customers c WHERE EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.customer_id);
SELECT * FROM customers c WHERE NOT EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.customer_id);

How they differ from an inner join

• An inner join returns one row per matching pair, so a customer with three orders appears three times; a semi join returns them once.
• Semi and anti joins return only left-side columns.
• Spark can stop looking after the first match, so they can be cheaper.

How Spark runs them Like other joins: a broadcast hash join when the right side is small, otherwise a shuffle (sort-merge or shuffled hash join). Adaptive query execution and broadcast hints apply as usual.

Typical uses

• Semi: keep only events for active customers; filter a large fact table by a list of keys.
• Anti: find new records not yet in the target table (incremental loads); find orphan records, such as orders whose customer doesn't exist; data quality checks between systems.

The NULL trap NOT IN behaves differently from NOT EXISTS and anti joins when the subquery contains NULL. WHERE customer_id NOT IN (SELECT customer_id FROM orders) returns no rows at all if any orders.customer_id is NULL, because comparison with NULL is unknown. A left anti join or NOT EXISTS returns the expected rows, so prefer them.

Other notes

• There are no right semi or anti joins in the DataFrame API; swap the tables.
• For "keep rows that match a small list", isin() on a literal list is simpler than a semi join.
• Anti joins are the standard way to check completeness between two systems: run one in each direction to list records missing on either side.

Lineage is the record of how each DataFrame or RDD was derived from its sources: the chain of transformations Spark would need to replay to rebuild it. The DAG (directed acyclic graph) is how Spark organises that chain for execution: operations are nodes, dependencies are edges, and the scheduler cuts the graph into stages at shuffle boundaries. Lineage is mainly about recovery; the DAG is mainly about execution.

Lineage

• Every transformation produces a new dataset that remembers its parent and the operation applied.
• If an executor is lost, Spark recomputes only the lost partitions by replaying their lineage from the source or from the last shuffle output that still exists.
• It needs no replication of intermediate data, which is why it is efficient.
• Very long lineages (iterative algorithms, loops that keep transforming the same DataFrame) make planning and recovery slow; checkpoint() or localCheckpoint() saves the data and truncates the lineage, as does writing an intermediate table.

You can see it with rdd.toDebugString() for RDDs, or the logical plan in df.explain(True) for DataFrames.

DAG

1. When an action runs, the DataFrame's optimised physical plan (or the RDD lineage) is handed to the DAG scheduler.
2. The scheduler walks the dependencies backward from the final result. Narrow dependencies are pipelined into the same stage; each wide dependency (a shuffle) starts a new stage.
3. Stages are submitted in dependency order; independent stages can run in parallel.
4. Each stage becomes a set of tasks, one per partition, handed to the task scheduler to run on executors.
5. If a shuffle's output is lost, the scheduler re-runs the parent stage that produced it.

Example

from pyspark.sql import functions as F

daily = (spark.read.parquet("/data/events")               # stage 1: scan
          .filter(F.col("type") == "purchase")             # stage 1 (narrow)
          .groupBy("event_date").agg(F.count("*")))        # shuffle → stage 2
customers = spark.read.parquet("/data/customers")          # can run in parallel

Seeing it The Spark UI shows a DAG visualisation for each job and stage, and the SQL tab shows the physical plan as a graph with metrics per operator, such as rows output, time and spill. It is the first place to look when a stage is slow.

Why both matter The DAG lets Spark plan the whole job, pipeline narrow operations and minimise shuffles. Lineage lets it recover from failures without the cost of replicating data. Together they give Spark fault-tolerant, optimised execution.

Avro is a row-based format built for writing and exchanging records one at a time, with a schema that travels with the data and strong rules for schema evolution. ORC is a columnar format built for analytical reads. Neither is better in general: Avro suits streaming, messaging and ingestion; ORC (or Parquet) suits analytics.

When Avro is preferable

• Streaming and messaging: each Kafka message is one Avro record, validated against a schema in a schema registry.
• Write-heavy landing of raw data, where whole records are always read back.
• Contracts between teams or systems where schemas evolve and compatibility rules matter.

When ORC is preferable

• Analytical queries that read a few columns and filter rows: columnar reading, compression and statistics make them far faster than Avro.
• Existing Hive ecosystems, where ORC is the native format and supports Hive's ACID tables.

Where Parquet fits In Spark-centred and cloud lakehouse systems, Parquet is the more common columnar choice: it has the widest tool support and is the storage format under Delta Lake and Iceberg (Iceberg can also use ORC and Avro). ORC and Parquet perform similarly; ecosystem support usually decides.

A common pattern Data arrives as Avro (or JSON) from Kafka or change data capture, is landed as is, then converted into columnar tables (Parquet or ORC, usually through Delta or Iceberg) for analytics. So the honest answer is: choose by workload, and often use both.

# Kafka value in Avro → columnar table
from pyspark.sql.avro.functions import from_avro

events = (spark.readStream.format("kafka")
            .option("kafka.bootstrap.servers", "broker:9092")
            .option("subscribe", "orders")
            .load()
            .select(from_avro("value", orders_avro_schema_json).alias("o"))
            .select("o.*"))

Use Spark's JDBC data source with the database's JDBC driver on the classpath. Read a table or a query into a DataFrame, in parallel by splitting on a numeric or date column, and write DataFrames back with write.jdbc or format("jdbc").

Read

orders = (spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://db.internal:5432/shop")
    .option("dbtable", "public.orders")            # or .option("query", "SELECT ...")
    .option("user", user)
    .option("password", secret_password)            # from a secret store, never in code
    .option("partitionColumn", "order_id")          # numeric, date or timestamp
    .option("lowerBound", "1")
    .option("upperBound", "50000000")
    .option("numPartitions", "16")                  # 16 parallel queries
    .option("fetchsize", "10000")
    .load())

Without partitionColumn, the whole table is read by a single task through one connection. The bounds only decide how the range is split into partitions; they don't filter rows.

Pushdown Filters, column selection, limits, top-N and simple aggregates are pushed into the SQL sent to the database where possible, so orders.filter("status = 'open'").select("order_id") reads only those rows and columns. Check with explain() (PushedFilters, PushedAggregates).

Write

(result.write.format("jdbc")
    .option("url", "jdbc:postgresql://db.internal:5432/shop")
    .option("dbtable", "reporting.daily_revenue")
    .option("user", user).option("password", secret_password)
    .option("batchsize", "10000")
    .option("truncate", "true")        # with overwrite: keep the table definition
    .mode("overwrite")
    .save())

Limit write parallelism (coalesce or numPartitions) so Spark doesn't open more connections than the database can handle. JDBC writes don't support upserts directly: write to a staging table and run a MERGE in the database, or use foreachPartition with your own statements.

SQL interface

CREATE TEMPORARY VIEW pg_orders
USING jdbc
OPTIONS (url 'jdbc:postgresql://db.internal:5432/shop', dbtable 'public.orders',
         user '...', password '...');
SELECT * FROM pg_orders WHERE status = 'open';

Good practice

• Read from a replica, not the production primary, and agree load limits with the database owners.
• Keep credentials in a secret manager.
• For regular large extracts, prefer change data capture or database exports to files over repeated full JDBC scans.
• Managed platforms offer catalog-level federation (for example Lakehouse Federation in Databricks) that registers the external database so it can be queried like other tables.

They are two front ends to the same engine. SQL queries and DataFrame operations are both turned into the same logical plan, optimised by Catalyst and executed identically, so performance is the same for the same logic. The differences are in how you write, compose, test and maintain the code.

Same plan, same speed

spark.sql("SELECT country, SUM(amount) AS revenue FROM orders WHERE amount > 0 GROUP BY country")

from pyspark.sql import functions as F
(spark.table("orders").filter(F.col("amount") > 0)
      .groupBy("country").agg(F.sum("amount").alias("revenue")))

explain() on both shows the same optimised plan.

When SQL fits better

• Analytical queries with several joins and aggregations, especially when analysts will read or change them.
• Logic shared with SQL-first tools: dbt models, BI tools, SQL warehouses.
• Window functions and set operations, which many people find clearer in SQL.

When the DataFrame API fits better

• Pipelines with steps that depend on configuration: columns chosen at runtime, transformations applied in a loop, conditional logic.
• Reusable, unit-tested transformation functions.
• Integration with Python libraries, UDFs and machine learning code.
• Avoiding SQL string concatenation, which is fragile and a security risk with user input.

Things to know

• Both support the same functions; SQL-only functions are reachable from DataFrames through F.expr() or selectExpr.
• Parameterised SQL (spark.sql("... WHERE country = :c", args={"c": "DE"})) keeps SQL safe when values come from code.
• Python UDFs, when needed, can be registered and called from SQL.

Most real projects mix the two; see the next question for how.

A DStream (discretised stream) is the abstraction of Spark's original streaming API: a continuous stream represented as a sequence of RDDs, one per batch interval. Spark Streaming collected incoming data for, say, 5 seconds, turned it into an RDD, ran RDD operations on it, and repeated. DStreams are now legacy: Spark's documentation says Spark Streaming is the previous generation of the engine, receives no more updates, and that new applications should use Structured Streaming.

How DStreams worked

from pyspark.streaming import StreamingContext

ssc = StreamingContext(spark.sparkContext, batchDuration=5)       # 5-second batches
lines = ssc.socketTextStream("localhost", 9999)
counts = (lines.flatMap(lambda line: line.split(" "))
               .map(lambda w: (w, 1))
               .reduceByKey(lambda a, b: a + b))
counts.pprint()
ssc.start()
ssc.awaitTermination()

• Input DStreams came from sources such as Kafka, Kinesis, sockets and files, many through receivers that ran as long-lived tasks on executors.
• Transformations mirrored RDD operations (map, filter, reduceByKey, join), applied to each batch's RDD.
• Window operations (window, reduceByKeyAndWindow) combined several batches, such as a 1-minute window sliding every 10 seconds.
• Stateful operations (updateStateByKey, mapWithState) carried state across batches.
• Output operations (foreachRDD, saveAsTextFiles) wrote results.
• Checkpointing stored metadata and state for recovery; write-ahead logs protected receiver data.

Their role, and their limits DStreams brought micro-batch streaming to Spark and unified it with batch processing on RDDs. But they:

• worked on processing time, not event time, so late or out-of-order data was hard to handle correctly;
• used RDDs, so no Catalyst optimisation;
• left exactly-once output to the developer;
• had a separate API from batch DataFrame code.

What replaced them Structured Streaming treats a stream as an unbounded table and uses the DataFrame and SQL API, with event-time windows, watermarks, built-in exactly-once guarantees with supported sinks, and optimisation. New integrations, such as the Kafka connector improvements and the newer stateful API transformWithState, are built only for Structured Streaming.

In an interview Explain what DStreams are, then say you would build new streaming jobs with Structured Streaming and migrate existing DStream jobs: map the batch interval to a trigger interval, updateStateByKey to transformWithState or a watermarked aggregation, and foreachRDD to foreachBatch.

A watermark tells Structured Streaming how late event data may arrive and still be counted. Spark tracks the latest event time it has seen, subtracts the delay you set, and treats that point as the watermark: windows and state older than it are finalised and dropped, and records older than it may be ignored. It is what keeps state from growing forever and what lets append mode emit final results.

from pyspark.sql import functions as F

clicks = spark.readStream.table("bronze.clicks")    # has event_time

per_window = (clicks
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "page")
    .count())

(per_window.writeStream
    .outputMode("append")                             # emit each window once, when final
    .option("checkpointLocation", "/chk/clicks_5min")
    .toTable("silver.clicks_5min"))

How it works

1. In each micro-batch, Spark finds the maximum event_time seen so far, say 12:30.
2. The watermark is that minus the delay: 12:20. It is updated between micro-batches.
3. A 5-minute window ending at or before 12:20 can no longer receive data, so Spark writes its final count (in append mode) and removes its state.
4. A record with event time 12:14 arriving now belongs to a closed window and is dropped.

The guarantee, precisely Data that is less late than the delay is guaranteed to be included. Data later than that may or may not be included: it is not guaranteed to be dropped straight away, only eventually.

Choosing the delay It's a trade-off between completeness and latency and state size. Base it on how late your data really arrives, for example the 99th percentile of the difference between event time and arrival time. Too short loses late events; too long holds state and delays final results.

Where watermarks are needed or useful

• Windowed aggregations in append mode, which can only emit a window once it is final.
• Stream–stream joins, to bound how long each side's rows are kept; outer joins require them.
• Deduplication: dropDuplicates with a watermark, or dropDuplicatesWithinWatermark (Spark 3.5 and later), keeps only recent keys in state.
• Custom stateful processing with transformWithState, which can use event-time timers.

Conditions

• The watermark column must be the event-time column used in the aggregation, and withWatermark must be called before the aggregation.
• Output mode must be append or update; complete mode keeps all state, so watermarks don't clean anything up.
• With several input streams, Spark by default uses the minimum of their watermarks (spark.sql.streaming.multipleWatermarkPolicy set to min), so the slowest stream decides.

Monitoring The query's progress (query.lastProgress) reports the current watermark and the number of rows dropped as too late, which tells you whether the delay is right.

Stateful processing means a streaming query remembers information across micro-batches, such as running counts, open sessions or the rows waiting for a join partner. In Structured Streaming, Spark keeps this state in a fault-tolerant state store on the executors, versioned and checkpointed with each micro-batch, and watermarks decide when old state can be removed.

Built-in stateful operations

• Aggregations: groupBy(...).count(), running totals, and event-time windows (tumbling, sliding and session windows with session_window).
• Stream–stream joins: each side's rows are kept until a matching row can no longer arrive.
• Deduplication: dropDuplicates or dropDuplicatesWithinWatermark keeps the keys already seen.

from pyspark.sql import functions as F

sessions = (events
    .withWatermark("event_time", "30 minutes")
    .groupBy(F.session_window("event_time", "10 minutes"), "user_id")
    .agg(F.count("*").alias("events")))

Custom state For logic the built-in operators can't express, such as alerting when a device's reading stays high for 3 readings in a row, use arbitrary stateful processing:

• transformWithState (Spark 4.0 and later; transformWithStateInPandas in Python) is the current API. It supports several state variables per key (value, list and map state), time-to-live for state, processing-time and event-time timers, and schema evolution of state.
• applyInPandasWithState (Python) and mapGroupsWithState / flatMapGroupsWithState (Scala, Java) are the older APIs, still supported.

Where the state lives

• The default state store keeps state in executor memory and writes versioned files to the checkpoint location.
• The RocksDB state store (spark.sql.streaming.stateStore.providerClass set to the RocksDB provider) keeps state on local disk with a memory cache, which handles much larger state with less garbage collection; it is the usual choice for production.
• Changelog checkpointing for RocksDB writes only the changes in each batch, which shortens checkpoint time.

Fault tolerance Each micro-batch's offsets and state version are written to the checkpoint. After a failure, the query restarts from the last committed batch, reloads the matching state version and reprocesses from the source, giving exactly-once results with replayable sources and idempotent sinks. The checkpoint must stay on reliable storage, and some changes to the query (such as changing grouping keys) are incompatible with existing state.

Keeping state under control

• Always use watermarks (or state time-to-live in transformWithState) so state for finished keys is removed.
• Watch state metrics in the query progress and the Structured Streaming UI: number of state rows and memory used.
• Partition state sensibly: the number of shuffle partitions is fixed for a query once it has a checkpoint.

The old DStream approach DStreams used updateStateByKey and mapWithState. They are legacy; new stateful jobs use Structured Streaming.

Read each source with its own reader and an explicit schema, map every source onto one target schema (same names, same types, a column for the source), then combine them with unionByName, using allowMissingColumns=True for columns that exist in only some sources.

from pyspark.sql import functions as F

csv_df = (spark.read.option("header", True)
            .schema("id INT, name STRING, amount DOUBLE")
            .csv("/landing/orders_csv/"))

json_df = (spark.read
            .schema("id LONG, customer_name STRING, amount DOUBLE, channel STRING")
            .json("/landing/orders_json/"))

parquet_df = spark.read.parquet("/landing/orders_parquet/")   # schema in the files

def to_target(df, source, renames=None):
    for old, new in (renames or {}).items():
        df = df.withColumnRenamed(old, new)
    return (df.withColumn("id", F.col("id").cast("long"))
              .withColumn("amount", F.col("amount").cast("decimal(12,2)"))
              .withColumn("source", F.lit(source)))

combined = (to_target(csv_df, "csv")
    .unionByName(to_target(json_df, "json", {"customer_name": "name"}), allowMissingColumns=True)
    .unionByName(to_target(parquet_df, "parquet"), allowMissingColumns=True))

Handling the schema differences

• Different names: rename to the target names (withColumnRenamed, or withColumnsRenamed with a dictionary) before the union.
• Missing columns: unionByName(..., allowMissingColumns=True) fills them with null.
• Different types: cast explicitly to the target type. unionByName will widen compatible types (int and long become long) but fails or gives surprising results for incompatible ones. Remember that ANSI mode is on by default since Spark 4, so a cast that can't succeed raises an error; use try_cast where bad values are expected.
• Column order: unionByName matches by name; plain union matches by position and silently mixes up columns.
• Nested fields: align structs field by field, or flatten them first.

Why explicit schemas Schema inference reads extra data, may guess differently from day to day (for example, a zip code 01234 inferred as the number 1234), and differs between formats. Explicit schemas make the union predictable.

Bad records With the default PERMISSIVE mode, a malformed CSV or JSON line becomes a row of nulls (a CSV line with too few fields counts as malformed). Add a _corrupt_record column to the schema to keep the raw text, and filter those rows into a quarantine table before the union.

Parquet files with different schemas Within one Parquet source, option("mergeSchema", "true") combines compatible schemas across files.

In production Land each source as is (with its own schema), then build the combined table in a separate step, ideally as a Delta or Iceberg table with schema enforcement, so a new column or changed type in one source is caught instead of silently changing the combined data. Auto Loader and declarative pipelines on Databricks add schema evolution and rescued-data columns for this.

Start by finding where the time goes: open the query in the Spark UI's SQL tab, see which join strategy was used and which stage is slow, and compare task durations and shuffle sizes. The cause is almost always one of: too much data being shuffled, the wrong join strategy, skewed keys, badly sized partitions, or spilling because of too little memory. Each has a specific fix.

1. Diagnose

• Plan: df.explain("formatted") and the SQL tab show BroadcastHashJoin, SortMergeJoin or ShuffledHashJoin, and whether filters were pushed to the scans.
• Stage details: for the slow stage, compare the median and maximum task duration and shuffle read size. A maximum far above the median means skew.
• Spill: "Spill (memory)" and "Spill (disk)" columns show memory pressure.
• Sizes: input rows and shuffle bytes for each side of the join.
• Key distribution: df.groupBy("customer_id").count().orderBy(F.desc("count")).show(10) reveals hot keys, including nulls or placeholder values such as 0 or "unknown".

2. Fixes by cause

Too much data shuffled

• Filter and select before the join; avoid UDFs that block pushdown.
• Pre-aggregate if you only need totals per key.
• Check the join condition: a missing or wrong key can multiply rows.

Wrong join strategy

• Broadcast the small side if it fits: big.join(F.broadcast(small), "key"), or raise spark.sql.autoBroadcastJoinThreshold (10 MB by default) carefully.
• Keep statistics current (ANALYZE TABLE ... COMPUTE STATISTICS) so the planner sees true sizes.
• AQE (on by default) converts to a broadcast join at runtime if a side turns out small.

Skew

• AQE's skew join handling splits partitions over 5 times the median and over 256 MB in sort-merge joins; check that it happened (the plan shows skewed partitions in AQEShuffleRead).
• Handle null or placeholder keys separately: filter them out, join the rest, then union back.
• Salting for extreme cases: add a random suffix to hot keys on the large side and replicate the matching rows on the small side.

Partition sizing

• Too few shuffle partitions give huge tasks that spill; too many give overhead. Set spark.sql.shuffle.partitions (or spark.sql.adaptive.coalescePartitions.initialPartitionNum) high enough for the data, aiming at roughly 100–200 MB per partition, and let AQE coalesce.

Memory and spill

• Fewer cores per executor or more memory per executor gives each task more room; more partitions makes each task smaller.

3. Structural fixes for joins that run repeatedly

• Store both tables bucketed or clustered on the join key so the shuffle can be skipped (bucketing or storage-partitioned joins with Iceberg), or use liquid clustering on Databricks to improve file skipping.
• Materialise a pre-joined or pre-aggregated table if many queries need the same join.

Example

from pyspark.sql import functions as F

orders = spark.table("sales.orders").filter("order_date >= '2026-01-01'").select("order_id", "customer_id", "amount")
customers = spark.table("crm.customers").select("customer_id", "segment")

joined = orders.join(F.broadcast(customers), "customer_id")   # if customers is small enough
joined.explain("formatted")

Change one thing at a time and compare the stage metrics, so you know which fix helped.

Spark's CSV and JSON readers have three modes for malformed records, set with the mode option: PERMISSIVE (the default) keeps the row, sets fields that can't be parsed to null and can save the raw text in a corrupt-record column; DROPMALFORMED silently drops malformed rows; FAILFAST stops the read with an error on the first one. Separate options handle whole files that can't be read.

PERMISSIVE (default)

orders = (spark.read
    .option("header", True)
    .option("mode", "PERMISSIVE")
    .option("columnNameOfCorruptRecord", "_corrupt_record")
    .schema("id INT, name STRING, amount DOUBLE, _corrupt_record STRING")
    .csv("/landing/orders/"))

orders.cache()      # needed before querying the corrupt column on its own
bad = orders.filter("_corrupt_record IS NOT NULL")
good = orders.filter("_corrupt_record IS NULL").drop("_corrupt_record")

With the rows 1,Ann,10, 2,Bo,oops and 3,Cy, row 1 is clean; row 2 gets amount null and its raw text in _corrupt_record; row 3 has too few fields, so it is also treated as malformed. Spark doesn't allow a query that references only the corrupt-record column straight from raw CSV or JSON files, so cache or save the parsed result first.

DROPMALFORMED Only well-formed rows are returned (row 1 above). Simple, but you lose the bad data and don't know how much was dropped, so use it only when that's acceptable and you count rows elsewhere.

FAILFAST The read fails with a FAILED_READ_FILE error naming the file. Good for tests and for sources that must always be clean.

Whole files that can't be read

• ignoreCorruptFiles (option or spark.sql.files.ignoreCorruptFiles): skip files that can't be read at all, such as truncated Parquet files, and return the rest.
• ignoreMissingFiles: skip files deleted after the file list was made. Both are off by default; turning them on hides problems, so log or monitor what is skipped.

Databricks: badRecordsPath On Databricks, .option("badRecordsPath", "/quarantine/orders") writes malformed records and unreadable files, with the reason, to that location as JSON files, and continues the read. Auto Loader adds a rescued-data column that captures values that don't match the schema.

Recommended practice

• Use PERMISSIVE with an explicit schema and a corrupt-record column in pipelines.
• Write bad rows to a quarantine table with the file name (_metadata.file_path) and load time.
• Alert on the bad-row count or rate, rather than only failing or silently dropping.
• Use FAILFAST in tests.
• Parquet and other self-describing formats don't use these modes, since rows can't be partially malformed; there, the issue is whole corrupt files.

The driver plans the work and collects results. It does not process the data, so a driver out of memory almost always means data was pulled to it on purpose or by accident: collect(), toPandas(), a broadcast that turned out larger than expected, or plain Python code working on large data. Find which line did it, then keep that work on the executors.

The usual causes

Fixes, one per cause

1. Results pulled to the driver. collect() and toPandas() load every row into the driver's memory. Write the result to a table or files instead, and read back only what a person needs to see. If you need a few rows, use limit() first.
2. An oversized broadcast. Spark estimates table sizes when it plans a broadcast, and the estimate can be wrong. Compute statistics on the table (ANALYZE TABLE), remove a broadcast() hint that no longer fits, or ask for a sort-merge join with a MERGE hint.
3. Python on the driver. A loop over rows, or pandas code on a large frame, runs on the driver alone, however large the cluster is. Rewrite it with DataFrame operations so the executors share the work.
4. A shared driver. Run scheduled jobs on their own job clusters, not on the interactive cluster everyone uses.

About spark.driver.maxResultSize

It caps the total size of results of one action (1 GB by default) and aborts the job above it. That error is a guard doing its job. Raising the cap without raising driver memory swaps a clear error for a crash. Raise it only when you know the result is meant to be that size.

When more memory really is the answer

If the driver still runs short once the code keeps the data on the executors, give it more: raise spark.driver.memory or pick a larger driver node. In client mode the setting has to be given when the application starts (--driver-memory), not from inside it, because the driver is already running by then.

What interviewers want: knowing that the driver does not process data, naming collect(), toPandas() and broadcasts as the main suspects, and fixing the code before resizing the driver.

Control how many tasks write and how the rows are spread across them, so each output file is a useful size, and compact what is already there. Many small files make every reader pay to list and open each one. The usual causes are writing straight after a shuffle with many partitions, partitioning the table by a column with too many values, or a streaming job that writes every few seconds.

Why it happens

Spark writes in parallel: every task writes its own files. The number of files is roughly the number of writing tasks times the number of table partitions each task's rows fall into.

• A write after a join or aggregation uses the shuffle's partition count, 200 by default, even for a small result.
• partitionBy("customer_id") on a column with 100,000 values creates folders that each hold a few rows, multiplied by the tasks that touched them.
• A stream that commits every 10 seconds writes new files every 10 seconds.

Fixes when writing

1. Size the write. Bring the data into fewer, fuller partitions before writing:

   (df.repartition("event_date")          # rows of one date go to one task
      .write.partitionBy("event_date")
      .format("delta").mode("append").save(path))
   

   For an unpartitioned table, coalesce(n) with n from the output size divided by the file size you want (Databricks aims optimized writes at 128 MB, a reasonable default). Repartitioning by the table's partition column means each folder gets files from one task, not from all of them.
2. Partition the table by a low-cardinality column. Partition by date, not by customer or ID. If you need fast lookups on a high-cardinality column, use clustering inside the files instead of folders.
3. Let the platform size files. On Delta tables in Databricks, optimized writes add a shuffle before the write to aim for about 128 MB files, and auto compaction merges small files after a write. Unity Catalog managed tables get much of this tuned for you.
4. Cap huge files too. maxRecordsPerFile stops a single task from writing one enormous file, which matters after a coalesce(1).
5. Streaming: use a longer trigger interval when the latency allows it, and compact the table on a schedule.

Fixes for files already written

• Delta Lake: run OPTIMIZE on the table, or on the recent partitions only. Large tables need it on a schedule, since auto compaction is not a full replacement.
• Plain Parquet: read the partition, rewrite it with the right number of files, and swap it in.
• The replaced small files stay in storage until VACUUM removes them, by default only once they are older than 7 days, so plan for the overlap.

Trade-offs to mention

A repartition before writing adds a shuffle, so writes take longer. That is usually worth it for a table read many times, and not for a scratch table read once.

What interviewers want: why the file count is tasks times partitions, a write-side fix (repartition by the partition column, sensible partitioning, optimized writes) and a read-side fix (compaction), with the trade-off stated.

Replace the UDF with built-in Spark functions if you can. If you cannot, turn it into a vectorized pandas UDF that works on whole batches. A plain Python UDF is slow because every row is sent from the JVM to a Python process, handled one at a time by your function, and sent back. Built-in functions run inside the JVM and skip that trip.

1. Confirm the UDF is the cost

In the Spark UI's SQL tab, the query plan shows a separate Python evaluation step for the UDF. Find the stage that runs it and compare its time with the rest of the job. A quick test is to run the job once with the UDF column replaced by a constant.

2. Rewrite it with built-in functions

Most UDFs in real code do things Spark already has: string cleanup, parsing dates, mapping codes, simple maths.

# Before: a Python UDF
@udf("string")
def clean_phone(s):
    return re.sub(r"[^0-9]", "", s) if s else None

# After: the same with a built-in, no Python involved
df = df.withColumn("phone", F.regexp_replace("phone", r"[^0-9]", ""))

Other common swaps: F.when chains for if/else logic, F.to_date with a format, a join to a small mapping table instead of a dictionary lookup, and higher-order functions (F.transform, F.filter, F.aggregate) for arrays.

3. If the logic really needs Python, vectorize it

A pandas UDF receives a batch of rows as a pandas Series, transferred with Arrow, so your code can use vectorized operations:

@pandas_udf("double")
def score(amount: pd.Series, days: pd.Series) -> pd.Series:
    return np.log1p(amount) / (days + 1)

This is the right choice for numeric work with NumPy, or for calling a library (a tokenizer, a model) on a batch at a time.

4. If you must keep a row-at-a-time UDF

Use Arrow for the transfer: @udf(..., useArrow=True), or the session setting spark.sql.execution.pythonUDF.arrow.enabled. From Spark 4.2 this is on by default for regular Python UDFs. The function still runs once per row, so it helps less than a pandas UDF.

Other things to check

• Calling the UDF on more rows than needed. Filter first, then apply it.
• Expensive setup inside the function. Loading a model or compiling a pattern on every call. Do it once per batch (pandas UDF) or once per worker.

What interviewers want: why Python UDFs are slow (row by row, data moved between the JVM and Python), the order of fixes (built-ins, then pandas UDFs, then Arrow), and an example of a real rewrite.

A join multiplies rows when a key appears more than once on both sides: a key with 3 rows on the left and 4 on the right produces 12 rows. So the cause is almost always a key that is not unique where you assumed it was, or a join condition that is missing part of the key. Count rows per key on each side, find the keys that repeat, and fix either the data or the join.

1. Measure the fan-out

left_dups = left.groupBy("customer_id").count().filter("count > 1")
right_dups = right.groupBy("customer_id").count().filter("count > 1")
left_dups.orderBy(F.desc("count")).show(10)
right_dups.orderBy(F.desc("count")).show(10)

In the Spark UI's SQL tab, the join operator's "number of output rows" against its inputs shows the multiplication at a glance, even in a long job.

2. The usual causes

3. Fixes

• History tables: join on the key and the validity period, or filter to the current row first:

  current = customers.filter("is_current = true")
  

• Missing key columns: add every column that makes the match unique.
• Duplicate loads: fix the load so it is safe to rerun, and remove the duplicates at the source, not after the join.
• Placeholder keys: treat them as missing. Join only real keys, and handle the rest separately.
• Facts to facts: aggregate one side to the grain of the other before joining.

Only use dropDuplicates() after the join if you know why the duplicates exist. Removing them blindly hides a data problem and can drop rows that are genuinely different.

4. The opposite surprise: missing rows

NULL keys never match with ==, because comparing NULL with anything gives NULL, not true. If rows vanish in an inner join, check for NULL keys. If NULL should match NULL, use eqNullSafe:

left.join(right, left.region.eqNullSafe(right.region))

5. Stop it reaching production

Add checks to the pipeline: assert that a dimension has one row per key before it is used in a join, and compare the joined row count with the count of the driving table. Fail the job when the ratio jumps.

What interviewers want: the multiplication explained (3 times 4 is 12), counting rows per key on both sides, the common causes with fixes, and an automated check so the next one is caught before anyone sees the report.

I start by measuring: find the slow stages in the Spark UI, see whether time goes to reading, shuffling, skew, spilling or Python code, and fix the biggest cause first. Most gains come from reading less data, shuffling less, and keeping work in Spark's optimised engine; configuration tuning comes after that.

1. Measure first

• In the Spark UI, find the stages that take the most time, and look at task time distribution (one long task means skew), shuffle read and write sizes, spill to disk, and garbage collection time.
• df.explain("formatted") shows the physical plan: join types, whether filters are pushed to the scan, and how many partitions are read.

2. Read less data

• Columnar formats (Parquet, Delta, Iceberg) so only needed columns are read.
• Filter early on partition columns and on columns with file statistics, so files are skipped.
• Fix many small files: compact them (OPTIMIZE in Delta) or write fewer, larger files.

3. Shuffle less, and shuffle well

• Broadcast small tables in joins (automatic under 10 MB by default, or with broadcast()).
• Aggregate or filter before joins and before groupBy.
• Keep adaptive query execution on (the default): it merges small shuffle partitions, switches joins to broadcast at runtime and splits skewed partitions.
• Set spark.sql.shuffle.partitions sensibly for very large jobs, aiming at roughly 100–200 MB per partition.

4. Handle skew AQE's skew join handling covers most joins; for skewed aggregations or extreme cases, salt the hot keys or process them separately.

5. Stay in the optimised engine

• Use built-in functions instead of Python UDFs; when a UDF is unavoidable, use a vectorised (pandas or Arrow) UDF.
• Avoid collect() and toPandas() on large data.

6. Cache with purpose Cache only a DataFrame that is reused several times and expensive to compute, and unpersist() it afterwards.

7. Size the resources

• Executors with about 4–5 cores each and enough memory per core to avoid spills; leave room for memory overhead, which Python processes use.
• Dynamic allocation or autoscaling for variable workloads.
• Check that the cluster actually is busy: idle cores mean too few partitions.

8. Write efficiently Partition output tables only by low-cardinality columns, avoid producing thousands of tiny files, and use table features such as clustering where the platform supports them.

9. Change one thing at a time Re-run with realistic data, compare stage times, and keep the change only if it helps.

Data skew means a few keys or partitions hold far more data than the rest, so a few tasks run much longer than the others and the stage waits for them. Start by confirming and locating it, then use adaptive query execution's skew handling, broadcast joins, salting or separate handling of the hot keys.

1. Detect it

• Spark UI: in the slow stage, compare the maximum task duration and shuffle read size with the median. A maximum many times the median means skew.
• Find the hot keys: df.groupBy("key").count().orderBy(F.desc("count")).show(20).
• Common causes: NULL or default keys ("unknown", 0), a few very large customers, a single date with a backlog.

2. Let adaptive query execution handle skewed joins AQE is on by default. In sort-merge and shuffled hash joins it splits a partition that is larger than 5 times the median and over 256 MB (spark.sql.adaptive.skewJoin.*) into smaller tasks, replicating the matching rows from the other side. This fixes many skewed joins with no code change. Check the plan for skew=true in the SQL tab.

3. Broadcast the small side If one side of the join is small, broadcasting it avoids the shuffle, and with it the skew:

from pyspark.sql.functions import broadcast
result = facts.join(broadcast(dim), "customer_id")

4. Deal with NULL and junk keys Rows with NULL keys never match in an inner join, so filter them out before the join; or, for an outer join, give each a random key so they spread out.

5. Salt the hot keys (aggregations) Add a random salt, aggregate on key plus salt, then aggregate again on the key:

from pyspark.sql import functions as F

SALT = 16
partial = (df.withColumn("salt", (F.rand() * SALT).cast("int"))
             .groupBy("key", "salt").agg(F.sum("value").alias("partial_sum")))
result = partial.groupBy("key").agg(F.sum("partial_sum").alias("total"))

This works for sums, counts, minimums and maximums; for averages, carry sums and counts and divide at the end.

6. Salt joins when AQE isn't enough Salt the large side randomly and replicate the small side once per salt value:

facts_salted = facts.withColumn("salt", (F.rand() * SALT).cast("int"))
dim_salted = dim.crossJoin(spark.range(SALT).withColumnRenamed("id", "salt"))
joined = facts_salted.join(dim_salted, ["key", "salt"]).drop("salt")

It multiplies the small side by the salt count, so salt only the hot keys when the other side is large.

7. Split hot keys out Process the few hot keys separately (often with a broadcast join), process the rest normally, and union the results.

8. Fix it at the source Pre-aggregate data upstream, choose partition columns with even distribution, and avoid writing tables partitioned by a skewed column.

Re-check the stage's task-time distribution after each change: skew fixes add complexity, so keep only what makes a measurable difference.

Memory problems in PySpark usually come from one of four places: pulling too much data to the driver, partitions too large for an executor's memory (often because of skew), Python processes using memory outside the JVM that the container doesn't allow for, or caching more than fits. The fix depends on which one it is, so read the error first.

Recognising the failure

Causes and fixes

1. Data to the driver Avoid collect() and toPandas() on large data. Use show(), take(), aggregate first, or write results to storage. Raise spark.driver.maxResultSize only if the result really is meant to be large.

2. Partitions too large Each task holds its partition's working data. Increase the number of shuffle partitions so each is about 100–200 MB, and let adaptive query execution coalesce small ones. For file reads, spark.sql.files.maxPartitionBytes (128 MB by default) controls split size.

3. Skew One huge partition runs out of memory while the others are fine. AQE skew join handling, broadcast joins and salting spread the data.

4. Python memory Python UDFs and pandas UDFs run in Python worker processes outside the JVM heap. Increase spark.executor.memoryOverhead (by default 10% of executor memory, at least 384 MB) or set spark.executor.pyspark.memory, and reduce Arrow batch sizes. Better: replace Python UDFs with built-in functions.

5. Aggregations that build big collections collect_list, collect_set and RDD groupByKey gather all values of a key into memory. Use aggregations that combine as they go (sum, count, approx_count_distinct), or reduceByKey instead of groupByKey on RDDs.

6. Too much caching Cache only what is reused, and unpersist() afterwards. A DataFrame's cache() uses MEMORY_AND_DISK, so partitions that don't fit go to disk rather than failing, but cached data still competes with execution memory.

7. Large broadcasts Broadcasting a table that is too large can exhaust driver and executor memory; check sizes before forcing broadcast().

8. Executor sizing Raise spark.executor.memory, or keep memory the same and lower spark.executor.cores, so each task gets more memory. Very large heaps (well over 64 GB) suffer long garbage collection pauses; more, medium-sized executors are usually better.

spark.memory.fraction (0.6 by default) rarely needs changing; partition sizing and avoiding driver collection solve most cases.

Dynamic partition pruning (DPP) lets Spark skip partitions of a large partitioned table using values that are only known when the query runs, typically the result of a filter on a small table it is joined to. Without it, a filter on the dimension table doesn't reduce what is read from the fact table, because the partition values aren't known when the query is planned.

The problem it solves

SELECT d.quarter, SUM(s.amount)
FROM sales s                      -- huge, partitioned by date_id
JOIN dates d ON s.date_id = d.date_id
WHERE d.quarter = 'Q1'            -- filter only on the small table
GROUP BY d.quarter;

Static partition pruning only works when the filter is on the partition column itself, such as WHERE s.date_id BETWEEN .... Here the filter is on dates, so a plan without DPP reads every partition of sales and throws most rows away in the join.

How DPP fixes it

1. Spark evaluates the filtered dates side first.
2. It takes the distinct date_id values that survive (those in Q1).
3. It applies them as a partition filter on the sales scan, so only matching partition directories are listed and read.

When the small side is broadcast for the join, Spark reuses the same broadcast result for the pruning, so the extra work is almost free. In the physical plan you see a SubqueryBroadcast dynamicpruning node and, on the fact table's scan, PartitionFilters: [... dynamicpruningexpression(date_id IN dynamicpruning#..)].

Benefits

• Far less data read and fewer tasks: often a large fraction of a fact table's partitions are skipped in star-schema queries.
• Less shuffle and memory downstream, since fewer rows reach the join.
• No code change: it is on by default (spark.sql.optimizer.dynamicPartitionPruning.enabled) since Spark 3.0.

When it applies

• The large table is partitioned by the join key.
• The join is an equi-join.
• The other side has a selective filter.
• The join type allows pruning that side (inner joins and left semi joins are the common cases).
• With default settings, the filtering side is broadcast so its result can be reused.

Related techniques For tables that aren't partitioned by the join key, Spark's runtime Bloom filters reduce the rows read from the other side of a shuffle join, and Delta Lake on Databricks adds dynamic file pruning, which skips individual files using their column statistics.

The usual challenge is that DPP silently doesn't happen: the table isn't partitioned by the join key, the join isn't eligible, or the small side isn't broadcast. The other challenges are designing partitions that make it useful and verifying it in the plan.

1. Table layout

• DPP only prunes partition columns. If sales is partitioned by region but joined on date_id, nothing is pruned.
• Partitioning by a high-cardinality key (such as customer_id) creates huge numbers of small files, which costs more than DPP saves. Pick partition columns that are both commonly joined or filtered on and low in cardinality, such as a date.

2. The small side isn't broadcast With default settings, Spark only uses DPP when it can reuse the broadcast of the filtering side. If that side is too big for the broadcast threshold (10 MB by default), or broadcast is disabled, the plan shows dynamicpruningexpression(true), meaning pruning was switched off. Filtering the dimension more, raising the threshold modestly, or a broadcast hint can bring it back.

3. Join type and conditions

• Only equi-joins on the partition column qualify; a join on an expression of the column (CAST(s.date_id AS STRING) = d.key) or a non-equality condition doesn't.
• Outer joins only allow pruning the side whose unmatched rows are discarded. In sales RIGHT JOIN dates, the broadcastable side would have to be sales, so dates can't be reused and DPP doesn't happen.

4. Filter selectivity If the dimension filter keeps most rows, pruning saves little. Spark estimates whether the filter is selective enough when it would need a separate subquery.

5. Source support DPP works with file-based tables (Parquet, ORC, Delta) and data sources that support runtime filtering, such as Apache Iceberg. Other connectors, such as JDBC sources, don't benefit.

6. Verifying it

• explain("formatted") shows dynamicpruningexpression(... IN dynamicpruning#...) on the scan when it applies.
• The SQL tab of the Spark UI shows the number of partitions and files read by the scan; compare it with the total.

7. Old statistics Wrong size estimates can stop Spark broadcasting the small side. ANALYZE TABLE ... COMPUTE STATISTICS keeps estimates accurate, and adaptive query execution can switch to a broadcast at runtime when the filtered side turns out small.

Internally, DPP is an optimizer rule that inserts a runtime partition filter on the large table, built from the join key values of the filtered small table, and a planner step that decides how to compute those values, normally by reusing the broadcast built for the join.

1. Logical optimisation The rule PartitionPruning looks for joins where one side is a partitioned table scanned with the join key as a partition column, and the other side has a filter. It adds a DynamicPruningSubquery predicate to the partitioned side, meaning "partition key IN (join keys of the filtered other side)".

2. Is it worth it? If the filtering side won't be broadcast, Spark would have to run that side twice, once for pruning and once for the join. It only does that when its statistics suggest the filter removes enough data; by default it prefers to apply DPP only when it can reuse a broadcast (spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly, an internal setting that defaults to true).

3. Physical planning

• If the join is a broadcast hash join with the filtered side as the build side, the predicate becomes a DynamicPruningExpression whose values come from a SubqueryBroadcast reusing that same broadcast exchange.
• If no broadcast can be reused (and the subquery isn't worth running), the predicate is replaced with true: pruning is off, and the plan shows dynamicpruningexpression(true).
• With adaptive query execution, the final decision is made when the join strategy is known at runtime.

4. Execution

1. The broadcast side runs first: the dimension table is scanned, filtered and broadcast.
2. The SubqueryBroadcast extracts the distinct join keys from it.
3. The fact table's file index applies them as partition filters, listing and reading only matching partition directories.
4. The join then runs on the reduced data, using the same broadcast hash table.

What you see in a plan

FileScan parquet sales[amount, date_id]
  PartitionFilters: [isnotnull(date_id), dynamicpruningexpression(date_id IN dynamicpruning#16)]
  ...
  +- SubqueryBroadcast dynamicpruning#16, [date_id]
     +- BroadcastExchange ...
        +- Filter (quarter = Q1)
           +- FileScan parquet dates[...]

Related runtime filters For non-partition columns, Spark can inject a Bloom filter or a semi-join filter into the other side of a shuffle join (spark.sql.optimizer.runtime.bloomFilter.enabled), which reduces rows rather than skipping partitions.

DPP is on by default and needs no setting. The settings that matter are the one that switches it on, the broadcast thresholds that decide whether the filtering side can be broadcast and reused, adaptive query execution, and statistics that make those size estimates accurate. Most tuning is really about table layout and joins.

1. The switch

spark.conf.get("spark.sql.optimizer.dynamicPartitionPruning.enabled")   # 'true' by default

Turn it off only to diagnose a problem.

2. Broadcast thresholds

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)      # default 10 MB
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", 50 * 1024 * 1024)  # runtime decision under AQE

DPP normally reuses the broadcast of the filtered dimension table. If that table is above the threshold, there is no broadcast to reuse and no pruning. Raising the threshold moderately can enable DPP, but each broadcast table must fit in driver and executor memory. A broadcast() hint on the dimension does the same for one query.

3. Adaptive query execution spark.sql.adaptive.enabled is true by default. AQE can turn a sort-merge join into a broadcast join after the filtered dimension turns out small, and DPP then follows the final join strategy.

4. Internal settings (rarely changed)

• spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly (default true): only prune when a broadcast can be reused. Setting it to false lets Spark run the dimension side as a separate subquery when it estimates the saving is worth computing it twice.
• spark.sql.optimizer.dynamicPartitionPruning.useStats and fallbackFilterRatio (0.5): how that estimate is made with and without statistics. These are internal and may change between versions, so test any change on your workload.

5. Statistics

ANALYZE TABLE dates COMPUTE STATISTICS;
ANALYZE TABLE sales COMPUTE STATISTICS FOR COLUMNS date_id;

Accurate sizes let the optimizer choose broadcasts and estimate filter selectivity correctly.

6. Table design (the biggest factor)

• Partition fact tables by the column they are commonly joined and filtered on, usually a date, with modest cardinality.
• Join on the partition column directly, with matching types, so no cast hides it.

Nuances

• More broadcasting increases driver and executor memory use; watch for out-of-memory errors and broadcast timeouts.
• DPP helps file-based tables and sources that support runtime filtering, not, for example, JDBC sources.
• On Databricks, Delta tables also get dynamic file pruning, which skips files using statistics even on non-partition columns, with its own settings.
• Check every change with explain("formatted") (look for dynamicpruningexpression(... IN dynamicpruning#...) rather than dynamicpruningexpression(true)) and the partitions-read metric in the SQL tab.

DPP only prunes partition columns, generally needs the filtering side to be broadcast, works only for certain join shapes, and only helps when the filter is selective and the data source supports it. Each limitation has a workaround, but often the real answer is a better table layout or a different skipping technique.

Explicit alternative when DPP doesn't apply Compute the keys yourself and filter the fact table on the partition column; static partition pruning then applies:

from pyspark.sql import functions as F

q1_dates = [r.date_id for r in dates.filter("quarter = 'Q1'").select("date_id").collect()]
sales_q1 = sales.filter(F.col("date_id").isin(q1_dates))

Only do this when the key list is small.

Checking it worked explain("formatted") shows dynamicpruningexpression(... IN dynamicpruning#...) when DPP applies, and dynamicpruningexpression(true) when Spark decided not to prune. Compare partitions read with the total in the Spark UI's SQL tab.

The overall lesson: DPP is a free bonus when tables are partitioned well and joins follow a star schema; it isn't a substitute for a good layout.

Catalyst is Spark SQL's query optimizer. Every DataFrame operation and SQL query becomes a tree of operators that Catalyst analyses, rewrites with optimisation rules and turns into a physical plan, choosing join strategies and pushing work down to the data source. Because of it, the same query is fast whether written in SQL or the DataFrame API, in Python or Scala.

The four phases

1. Analysis: the unresolved logical plan (column and table names as written) is resolved against the catalog: names checked, types assigned, functions found. Errors like an unknown column are raised here.
2. Logical optimisation: rule-based rewrites of the logical plan, such as:
   • predicate pushdown: filters moved as close to the source as possible;
   • column pruning: only needed columns kept;
   • constant folding: 1 + 2 computed once;
   • simplifying expressions and removing redundant operations;
   • combining consecutive filters and projections;
   • rewriting some subqueries as joins.
3. Physical planning: one or more physical plans are generated and one is chosen, using statistics for cost-based decisions such as broadcast hash join versus sort-merge join, and join order when the cost-based optimizer is enabled.
4. Code generation: whole-stage code generation (Tungsten) compiles chains of operators into Java bytecode working on compact binary rows, avoiding per-row function calls.

Runtime re-optimisation Adaptive query execution extends Catalyst at runtime: after each shuffle stage it re-plans with actual sizes, coalescing small partitions, switching to broadcast joins and splitting skewed partitions. Dynamic partition pruning and runtime Bloom filters also use values only known while the query runs.

Seeing it work

df = (spark.read.parquet("/data/orders")
        .filter("country = 'DE'")
        .select("order_id", "amount"))
df.explain(mode="extended")   # parsed, analysed, optimised and physical plans

The optimised plan shows the filter pushed into the Parquet scan (PushedFilters) and only two columns read.

What it means for your code

• Write clear, declarative code; Catalyst handles ordering of filters and projections.
• Python UDFs are opaque: Catalyst can't push filters through them or optimise inside them, and data must move to Python. Prefer built-in functions.
• RDD operations bypass Catalyst entirely.
• Keep table statistics current (ANALYZE TABLE) so cost-based choices are good.
• Catalyst is extensible: platforms and libraries add their own rules, such as Delta Lake's.

With inferSchema enabled, Spark reads the CSV data (all rows, or a sample set by samplingRatio) before the real read, and for each column picks the narrowest type that fits every sampled value: integer, long, decimal or double, boolean, date, timestamp, otherwise string. Without inferSchema, every column is a string. Column names come from the header, or _c0, _c1 and so on without one.

Downsides

1. An extra pass over the data Inference reads the files before processing them, which can double the read time on large data.

2. Wrong types

• Codes with leading zeros become numbers: zip code 01234 is read as the integer 1234, and the zero is gone.
• Money becomes double, which introduces rounding; financial data needs decimal.
• Dates in non-default formats stay strings unless you set dateFormat or timestampFormat.
• Large IDs can be read as numbers when they should be text.

3. Sampling can miss values With samplingRatio below 1, a column may look like integers in the sample and contain text elsewhere. Under ANSI mode (the default in Spark 4), or with mode set to FAILFAST, the job then fails, or in the default PERMISSIVE mode, the offending values become NULL.

4. Types change between runs A new file with one text value in a numeric column changes the inferred type, so downstream code and table writes break unpredictably.

5. Weak handling of bad data With inference there is no agreed contract to check rows against, so malformed records are harder to detect.

Better practice

schema = "order_id INT, order_date DATE, amount DECIMAL(10,2), zip STRING"
df = (spark.read
        .option("header", True)
        .option("mode", "PERMISSIVE")
        .schema(schema)
        .csv("/data/orders/"))

• Define the schema explicitly in pipelines; it is also faster.
• Infer once while exploring, then copy the result (df.schema.simpleString()) into the code and correct it.
• For changing sources, tools such as Databricks Auto Loader store an inferred schema and evolve it in a controlled way, with unexpected values kept in a rescued data column.

I treat partitioning on two levels: how data is split into Spark partitions while a job runs, which decides parallelism and memory per task, and how tables are laid out on storage, which decides what later queries can skip. The goal for both is evenly sized pieces of a sensible size, aligned with how the data is filtered and joined.

1. In-memory partitions during processing

Size: aim for roughly 100–200 MB per partition after a shuffle and around 128 MB per input split. Too few partitions leave cores idle and risk out-of-memory errors; too many add scheduling overhead and tiny files.

Input: Spark splits files according to spark.sql.files.maxPartitionBytes (128 MB by default) and packs small files together. Splittable formats (Parquet, ORC, uncompressed or bzip2 CSV) allow this; a single gzip file is read by one task.

Shuffles: spark.sql.shuffle.partitions (200 by default) sets the number of partitions after joins and aggregations. Adaptive query execution then merges small partitions toward a 64 MB target, so for most jobs I set a generous starting number for large data and let AQE coalesce.

Explicit changes:

• repartition(n, "key") to spread data evenly or co-locate by key before heavy processing (it shuffles).
• coalesce(n) to reduce partitions without a shuffle, for example before writing fewer files.
• repartition("date") before a partitioned write so each task writes to few directories.

Parallelism check: the number of tasks per stage should be at least two to three times the total number of cores.

2. On-disk table layout

Partition columns: partition large tables by a column that most queries filter on and that has low cardinality, typically a date or month. Each partition should hold at least about 1 GB; partitioning by customer or by hour on modest data creates millions of small files.

File sizes: keep files in the range of about 128 MB to 1 GB. Compact small files (OPTIMIZE in Delta), and avoid writing many tiny files from too many partitions.

Beyond partitioning: for filters and joins on other columns, use clustering and data skipping: Z-ordering or liquid clustering in Delta, sort orders in Iceberg, or bucketing in Hive-style tables to avoid shuffles in repeated joins. Databricks now recommends liquid clustering, which replaces partitioning and Z-ordering, for all new tables.

3. Skew Check that partitions are even: one huge partition (a hot key or a very busy date) slows the whole stage. Use AQE skew handling, salting or separate processing for hot keys.

4. Verify

• In the Spark UI, task durations and input sizes within a stage should be similar.
• df.rdd.getNumPartitions() (classic Spark) shows the current number.
• After writing, check file counts and sizes per partition directory.

5. Example

(events
    .repartition("event_date")              # one shuffle, each task writes one date
    .write.mode("overwrite")
    .partitionBy("event_date")
    .option("maxRecordsPerFile", 5_000_000)
    .saveAsTable("silver.events"))

Partitioning is a design decision, so I revisit it when data volumes or query patterns change.

25 GB is moderate for Spark: a cluster of a few executors handles it easily, if the file can be split for parallel reading. So the first question is the file's format and compression; then define the schema, keep partitions around 128 MB, filter and project early, and write the result to a columnar format.

1. Can it be split?

• Uncompressed CSV or JSON Lines, Parquet, ORC, Avro: splittable. Spark reads 128 MB pieces in parallel (spark.sql.files.maxPartitionBytes), so 25 GB becomes about 200 tasks.
• Gzip-compressed CSV: not splittable. One task reads the whole file, slowly, while the rest of the cluster waits. Read it once, repartition immediately, and write it to Parquet or Delta; better, ask the producer for split files or a splittable compression such as bzip2, or zstd-compressed Parquet.
• Multiline JSON or CSV with line breaks inside quoted fields (multiLine true) also can't be split.

2. Define the schema Inference on 25 GB means an extra full pass. Provide the schema:

schema = "order_id BIGINT, customer_id BIGINT, order_ts TIMESTAMP, amount DECIMAL(12,2), country STRING"
df = (spark.read
        .option("header", True)
        .option("mode", "PERMISSIVE")
        .schema(schema)
        .csv("/landing/orders_big.csv"))

3. Size the cluster sensibly About 200 input tasks fit well on, for example, 5 executors with 4–5 cores each and 16–32 GB of memory; it doesn't need to fit in memory, since Spark streams through partitions. Enable autoscaling or dynamic allocation if available.

4. Reduce early Select only needed columns and filter rows as early as possible. With columnar formats these are pushed into the scan; with CSV they still reduce shuffle and memory later.

5. Shuffle settings For joins and aggregations, set spark.sql.shuffle.partitions to roughly the data size after filtering divided by 128–200 MB, and leave adaptive query execution on to coalesce small partitions and handle skew. Broadcast small lookup tables.

6. Avoid driver bottlenecks No collect() or toPandas() of the full data; inspect with show() or aggregates.

7. Cache only if reused If several outputs come from the same cleaned data, cache it (MEMORY_AND_DISK) or write it once as an intermediate table.

8. Write efficiently

(df.write
    .mode("overwrite")
    .partitionBy("order_date")      # only if queries filter on it and partitions stay large
    .format("delta")
    .saveAsTable("silver.orders"))

Columnar output is typically several times smaller than CSV and much faster to query. Avoid producing thousands of tiny files: repartition to a sensible number before writing.

9. Make it repeatable If the file arrives regularly, ingest it incrementally (Structured Streaming's file source or Auto Loader), keep bad records in a quarantine column or table, and monitor the stage times in the Spark UI.

It depends on where Spark runs. While the application runs, the Spark UI (port 4040 on the driver) shows jobs and each executor's stdout and stderr. After it finishes, the history server replays the event logs, and the raw logs are with the cluster manager: YARN log aggregation, Kubernetes pod logs, standalone worker directories, or the platform's own log delivery.

Kinds of logs

• Driver logs: your application's own logging, planning and scheduling messages, and most error messages with stack traces.
• Executor logs: task errors, out-of-memory errors, garbage collection and Python worker errors.
• Event logs: a structured record of jobs, stages, tasks and SQL plans, used by the history server to rebuild the UI after the application ends.

While the application is running The Spark UI's Executors tab links to each executor's stdout and stderr.

After it finishes: the history server Enable event logging and point the history server at the same directory:

spark.eventLog.enabled   true
spark.eventLog.dir       hdfs:///spark-logs        # or s3a://bucket/spark-logs
spark.history.fs.logDirectory  hdfs:///spark-logs

The history server (port 18080 by default) shows the same pages as the live UI.

By cluster manager

• YARN: with log aggregation (yarn.log-aggregation-enable), container logs are copied to HDFS when the application ends; read them with yarn logs -applicationId <application id>. Without it, they stay on each node in the NodeManager's log directory. In client mode the driver's output is on the machine that ran spark-submit.
• Kubernetes: driver and executors are pods; kubectl logs <driver-pod> while they exist. Executor pods are deleted when the application ends, so ship logs to a central system (such as Fluent Bit to Elasticsearch, Loki or a cloud logging service).
• Standalone: each worker writes executor logs to its work/<app id>/<executor id>/ directory; the master and worker web UIs link to them.
• Local mode: everything goes to the console.

Managed platforms

• Databricks (classic compute): driver logs and the Spark UI in the compute's details, and cluster log delivery to a storage location you configure. On serverless compute the Spark UI and Spark logs aren't available; use the query profile.
• Amazon EMR: logs archived to S3 when configured, and the persistent Spark history server in the console.
• Google Dataproc and Azure Synapse/HDInsight: logs in the cloud's logging service and the Spark history server link in the console.

Practical tips

• Configure logging (Log4j 2 in current Spark) to control levels; spark.sparkContext.setLogLevel("WARN") reduces noise in notebooks.
• Turn on structured logging (spark.log.structuredLogging.enabled, available since Spark 4) to get JSON logs that can be searched centrally.
• When a job fails, start from the driver log for the first error, then follow the failed task to its executor log; the real cause (such as an out-of-memory error or a Python traceback) is usually there.

The deploy mode decides where the driver runs. In client mode the driver runs inside the process that started the application, such as spark-submit on an edge node, a notebook or your laptop. In cluster mode the cluster manager starts the driver on a machine inside the cluster, and spark-submit can exit once the application is launched.

spark-submit --master yarn --deploy-mode client  job.py     # driver on this machine
spark-submit --master yarn --deploy-mode cluster job.py     # driver in a YARN container

Client mode in more detail

• The driver's resources are the client machine's; a large collect() or broadcast can exhaust a small edge node.
• Executors must be able to connect back to the client, which matters with firewalls and in Kubernetes.
• Good for seeing output immediately and for interactive tools such as pyspark and notebooks.

Cluster mode in more detail

• The driver gets resources from the cluster (--driver-memory, --driver-cores) and fails over with the cluster manager's support (for example YARN application attempts).
• Logs must be collected from the cluster (yarn logs -applicationId ..., or driver pod logs in Kubernetes).
• Application files and dependencies must be reachable from the cluster (uploaded with --py-files, --files or --archives, or in shared storage).

Cluster manager specifics

• YARN: in cluster mode the driver runs in the application master container.
• Kubernetes: cluster mode creates a driver pod, which creates executor pods; client mode needs the driver to be reachable from the pods.
• Standalone: cluster mode isn't supported for Python applications, so PySpark jobs on standalone clusters run in client mode.
• Local (--master local[*]): everything runs in one process; deploy mode doesn't apply.

Managed platforms On Databricks, EMR Serverless or Dataproc Serverless you don't choose: the platform runs the driver for you, and notebooks connect to it, effectively client-style interaction with cluster-side drivers.

How I choose Client mode for exploration and debugging close to the cluster; cluster mode for scheduled production jobs, so they don't depend on the submitting machine staying up.

Start from the node size: give each executor about 4–5 cores, divide each node's memory between the executors on it after leaving room for the operating system and for memory overhead, and set the number of executors from the cores you want in total. Then check the Spark UI for spills, garbage collection time and idle cores, and adjust.

A worked example Cluster: 10 worker nodes, each with 16 cores and 64 GB of memory.

1. Reserve for the OS and node services: about 1 core and a few GB per node, leaving 15 cores and about 60 GB.
2. Cores per executor: 5. More than about 5 tends to hurt I/O throughput and makes garbage collection pauses longer; 1 wastes memory on per-executor overhead.
3. Executors per node: 15 ÷ 5 = 3, so 30 in total. On YARN, leave one executor's worth of capacity for the application master, giving 29.
4. Memory per executor: 60 GB ÷ 3 = 20 GB per container. Memory overhead defaults to 10% of executor memory (at least 384 MB) and covers off-heap use; PySpark jobs with Python UDFs need more. So set about 18 GB of executor memory and 2 GB of overhead.

spark-submit \
  --num-executors 29 \
  --executor-cores 5 \
  --executor-memory 18g \
  --conf spark.executor.memoryOverhead=2g \
  --driver-memory 8g \
  job.py

How memory inside an executor is used After 300 MB of reserved memory, spark.memory.fraction (0.6) of the heap is shared by execution (shuffles, joins, sorts) and storage (cache), which borrow from each other; the rest is for user data structures and Spark's internal metadata. These defaults rarely need changing.

Dynamic allocation For variable workloads or shared clusters, let Spark add and remove executors:

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.maxExecutors=50
spark.dynamicAllocation.shuffleTracking.enabled=true    # or an external shuffle service

Match parallelism to cores With 145 cores (29 × 5), each stage should have at least 2–3 times that many tasks. Set spark.sql.shuffle.partitions accordingly for large jobs, and let adaptive query execution coalesce small partitions.

Read the symptoms

Driver sizing Usually 4–8 GB is enough; more only if the job collects results, broadcasts large tables or plans very large queries.

Managed platforms On Databricks, EMR or Dataproc, you mostly choose instance types and autoscaling, and the platform sets executor sizes; serverless options size them for you. The same principles still guide which instance types to choose: memory-optimised for joins and caching, compute-optimised for CPU-heavy transformations.

Spark SQL is the Spark module for structured data. It lets you query data with SQL or the DataFrame API, works out the schema of the data, and runs every query through the Catalyst optimizer and Tungsten execution engine. "Core Spark" usually means the original RDD API: distributed collections of arbitrary objects processed with functions, with no schema and no query optimisation.

What Spark SQL provides

• A SQL engine with ANSI SQL behaviour (on by default since Spark 4), window functions, common table expressions, and features such as MERGE with table formats that support it.
• A catalog of databases, tables, views and functions (the Hive metastore, Unity Catalog or another catalog).
• JDBC and ODBC access through the Thrift server, so BI tools can query Spark.
• The basis for most of the rest of Spark: Structured Streaming, MLlib's DataFrame APIs and the pandas API on Spark all run on Spark SQL.

Example of the same job

# Spark SQL (DataFrame API)
from pyspark.sql import functions as F
(spark.read.parquet("/data/orders")
    .filter(F.col("amount") > 0)
    .groupBy("country").agg(F.sum("amount").alias("revenue")))

-- Spark SQL (SQL)
SELECT country, SUM(amount) AS revenue
FROM parquet.`/data/orders`
WHERE amount > 0
GROUP BY country;

# Core Spark (RDD)
(sc.textFile("/data/orders.csv")
    .map(lambda line: line.split(","))
    .filter(lambda f: float(f[2]) > 0)
    .map(lambda f: (f[1], float(f[2])))
    .reduceByKey(lambda a, b: a + b))

The RDD version has to parse the file by hand, can't skip columns or rows at the source, and is not optimised.

How they relate Spark SQL runs on the core engine: DataFrame plans are compiled into RDD operations underneath, using the same scheduler, executors and fault tolerance. Today almost all new work uses Spark SQL; the RDD API remains for low-level control and legacy code.

Yes. Both APIs share one session, one catalog and one engine, so you can switch freely. A DataFrame becomes available to SQL through a temporary view (or by passing it straight into spark.sql), and the result of spark.sql() is an ordinary DataFrame you keep transforming with the API.

DataFrame → SQL

orders = spark.read.parquet("/data/orders")
orders.createOrReplaceTempView("orders")

revenue = spark.sql("""
    SELECT country, SUM(amount) AS revenue
    FROM orders
    GROUP BY country
""")

Since Spark 3.4, PySpark can also reference DataFrames directly, without a view:

spark.sql("SELECT * FROM {o} WHERE amount > {m}", o=orders, m=1000)

SQL → DataFrame

from pyspark.sql import functions as F

top = (revenue
        .filter(F.col("revenue") > 1_000_000)
        .orderBy(F.desc("revenue")))

SQL expressions inside the DataFrame API

orders.selectExpr("order_id", "amount * 1.19 AS gross").filter("gross > 100")
orders.withColumn("band", F.expr("CASE WHEN amount > 100 THEN 'high' ELSE 'low' END"))

Parameters from code into SQL

spark.sql("SELECT * FROM orders WHERE country = :c", args={"c": "DE"})

Named parameter markers are safer than building SQL with f-strings.

Functions across both A Python function registered with spark.udf.register("normalise_email", fn) can be called from SQL; SQL functions created with CREATE FUNCTION can be used from F.expr.

What carries over and what doesn't

• Temporary views belong to the session; global temporary views are shared across sessions of the application; tables in the catalog are shared with everyone.
• Python variables are not visible in SQL unless passed as parameters or DataFrame references.
• A view is a query, not stored data: each use recomputes it, unless you cache it.

Mixing is normal practice: SQL for the readable core of a query, the DataFrame API for dynamic and reusable parts, with the same performance either way.

Spark Streaming processes data continuously as it arrives, while Spark batch processing runs over a fixed, already-complete dataset and then stops. "Spark Streaming" originally meant the DStream API, which is now a legacy project with no further updates; modern Spark streaming uses Structured Streaming, which runs streams with the same DataFrame and SQL API as batch.

Batch

• Input is bounded: a set of files or a table snapshot.
• A job reads everything, computes and finishes.
• Latency: minutes to hours, depending on schedule.
• Typical uses: nightly loads, reports, model training, backfills.

Streaming

• Input is unbounded: Kafka topics, event hubs, files landing in a folder, change feeds from Delta tables.
• A query keeps running, processing new data in small increments (micro-batches, by default as soon as the previous one finishes, or on a fixed interval).
• Latency: typically from under a second to a few minutes.
• Needs state for aggregations and joins over time, event-time handling for late data, and checkpointing for recovery.
• Typical uses: real-time dashboards, alerting, fraud detection, continuous ingestion into a lakehouse.

Structured Streaming in one example

from pyspark.sql import functions as F

events = (spark.readStream.format("kafka")
            .option("kafka.bootstrap.servers", "broker:9092")
            .option("subscribe", "clicks").load()
            .select(F.from_json(F.col("value").cast("string"), click_schema).alias("c"))
            .select("c.*"))

per_minute = (events
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "1 minute"), "page")
    .count())

(per_minute.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/chk/clicks_per_minute")
    .toTable("gold.clicks_per_minute"))

The transformation code is the same as a batch query; only readStream and writeStream differ.

Key differences in practice

A useful middle ground trigger(availableNow=True) runs a streaming query that processes everything new since the last run and then stops. You get streaming's incremental processing and checkpointed progress with batch-style scheduling, which is how many ingestion pipelines run today.

Structured Streaming is Spark's current streaming engine. It treats a stream as a table that keeps growing, lets you write the same DataFrame or SQL query you would write for batch data, and runs it incrementally, updating results as new data arrives with exactly-once guarantees. DStreams, the older API, represented a stream as a sequence of RDDs and is now a legacy project.

The model Each new record is a new row appended to an unbounded input table. Your query defines a result table; on each trigger, Spark processes only the new rows, updates any state, and writes changes to the sink according to the output mode (append, update or complete).

from pyspark.sql import functions as F

orders = (spark.readStream.format("delta").table("bronze.orders"))

revenue = (orders
    .withWatermark("order_ts", "30 minutes")
    .groupBy(F.window("order_ts", "5 minutes"), "country")
    .agg(F.sum("amount").alias("revenue")))

(revenue.writeStream
    .outputMode("append")
    .option("checkpointLocation", "/chk/revenue_5min")
    .trigger(processingTime="1 minute")
    .toTable("gold.revenue_5min"))

Differences from DStreams

Why it matters in practice

• One codebase for batch and streaming logic.
• Correct results with late data, through watermarks.
• Fault tolerance and exactly-once writes without hand-written bookkeeping.
• Incremental pipelines can run continuously or on a schedule with availableNow, the basis of most lakehouse ingestion today (including Databricks Auto Loader and Lakeflow declarative pipelines).

Limitations to mention

• Some operations aren't supported on streams, such as most sorting outside aggregations, and some outer joins need watermarks.
• Micro-batch latency is typically from a few hundred milliseconds to seconds; very low latency needs special modes or a different engine.
• State must be sized and monitored; unbounded state grows forever without watermarks.

A shuffle is Spark's mechanism for redistributing data across partitions so that rows with the same key end up together. It happens whenever an operation needs data from many input partitions to produce one output partition, and it is usually the most expensive part of a job because it involves disk writes, network transfer, serialisation and sorting.

Operations that cause shuffles

• Aggregations by key: groupBy().agg(), reduceByKey, groupByKey, distinct.
• Joins, unless one side is broadcast or both sides are already partitioned the same way (bucketed tables, storage-partitioned joins).
• repartition, repartitionByRange, and orderBy / sort (global sorting).
• Window functions with partitionBy.
• Set operations such as intersect and except.

Operations like filter, select, withColumn and union don't shuffle; they work on each partition independently.

How a shuffle works

1. Map side: each task in the stage before the shuffle partitions its output by key (hash or range) into as many buckets as there are shuffle partitions, and writes them to local disk as shuffle files.
2. Stage boundary: the next stage starts only when the map side is complete.
3. Reduce side: each task fetches its bucket from every map task over the network and processes it.

In the Spark UI, a shuffle appears as an Exchange in the SQL plan and as shuffle write and shuffle read sizes on the stages.

Why shuffles are costly

• Data is serialised and written to disk, then read and sent over the network.
• The number of files can be very large: map tasks × reduce partitions.
• Uneven keys cause skew: one reduce task gets far more data than the others.
• Memory pressure leads to spilling during sorts and aggregations.

How to reduce their cost

• Filter and select early so less data is shuffled; Catalyst does much of this automatically.
• Broadcast small tables in joins (automatic below spark.sql.autoBroadcastJoinThreshold, 10 MB by default, or with a broadcast hint).
• Prefer aggregations that combine on the map side: DataFrame aggregations do this automatically; with RDDs, use reduceByKey instead of groupByKey.
• Right-size shuffle partitions: spark.sql.shuffle.partitions defaults to 200. Adaptive query execution (on by default) merges small partitions after the shuffle, so set the initial number high enough for large jobs.
• Handle skew: AQE splits skewed partitions in sort-merge joins; for aggregations, salting the key helps.
• Reuse layouts: bucketed tables, or tables clustered on join keys, avoid repeated shuffles; caching a shuffled result helps when it is reused several times.
• Avoid unnecessary global sorts: sortWithinPartitions when total order isn't needed.

Infrastructure Shuffle files live on executors' local disks, so fast local SSDs help. Dynamic allocation needs shuffle tracking or an external shuffle service so shuffle files outlive executors, and some platforms add remote shuffle services for very large jobs.

Traditional query optimisation in Spark plans the whole query once, before it runs, using estimates: catalog statistics, file sizes and rules. Adaptive query execution (AQE) re-optimises the remaining plan while the query runs, using the actual sizes of the data produced by each completed stage. Since Spark 3.2, AQE is on by default.

Traditional, static optimisation

• Catalyst applies rules (predicate pushdown, column pruning, constant folding) and, where statistics exist, cost-based choices such as join order and join strategy.
• Every decision is fixed before execution: the number of shuffle partitions, whether a join is broadcast, how partitions are sized.
• Estimates are often wrong: statistics are missing or stale, filters are more or less selective than assumed, and sizes after joins and user-defined functions are guesses.

Adaptive query execution AQE splits the query at shuffle and broadcast exchanges into query stages. When a stage finishes, Spark knows the real size of every output partition, and re-optimises what is left:

• Coalescing shuffle partitions: merges many small post-shuffle partitions, so spark.sql.shuffle.partitions doesn't have to be tuned for each query.
• Switching join strategies: converts a sort-merge join to a broadcast hash join if one side turns out to be small, reading shuffle files locally to save network traffic.
• Handling skew: splits oversized partitions in sort-merge joins into smaller tasks.
• Choosing shuffled hash joins when partitions are small enough (configurable).

They work together AQE doesn't replace Catalyst: the static optimiser still produces the initial plan, and AQE adjusts it. Good statistics (ANALYZE TABLE, file-level statistics in table formats) still help the initial plan; AQE corrects what estimates got wrong.

Seeing it explain() shows AdaptiveSparkPlan isFinalPlan=false before execution; after the query runs, the SQL tab of the Spark UI shows the final plan, with nodes such as AQEShuffleRead (coalesced or skewed partitions) and any changed join strategies.

AQE applies five main optimisations once it knows the real sizes of shuffle and broadcast output: it coalesces small shuffle partitions, turns sort-merge joins into broadcast hash joins, turns sort-merge joins into shuffled hash joins, splits skewed partitions in joins, and splits skewed partitions when rebalancing. Each can be turned on or off separately.

1. Coalescing post-shuffle partitions After a shuffle, AQE merges neighbouring small partitions into larger ones.

• On by default (spark.sql.adaptive.coalescePartitions.enabled).
• Target size spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB). By default parallelismFirst is true, so Spark ignores the target and only respects a 1 MB minimum, keeping parallelism high; set it to false on a busy cluster.
• Set a large spark.sql.adaptive.coalescePartitions.initialPartitionNum and let AQE reduce it. Example: 200 shuffle partitions of 1 MB each become a handful of tasks instead of 200 tiny ones.

2. Converting sort-merge join to broadcast hash join If one side of a join turns out to be smaller than spark.sql.adaptive.autoBroadcastJoinThreshold (the same as the 10 MB static threshold by default), AQE switches to a broadcast join. The shuffle has already happened, so it isn't as efficient as planning a broadcast join from the start, but it avoids sorting both sides, and with the local shuffle reader (spark.sql.adaptive.localShuffleReader.enabled, on by default) executors read their own shuffle files instead of fetching over the network.

3. Converting sort-merge join to shuffled hash join If every post-shuffle partition is smaller than spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold, AQE uses a shuffled hash join, which avoids sorting. The threshold is 0 by default, so this is off until you set it at or above the advisory partition size.

4. Optimising skewed joins For sort-merge joins, a partition counts as skewed if it is more than skewedPartitionFactor (5) times the median size and larger than skewedPartitionThresholdInBytes (256 MB). AQE splits it into smaller tasks and replicates the matching partition on the other side. spark.sql.adaptive.forceOptimizeSkewedJoin applies this even when it adds an extra shuffle.

5. Splitting skew in rebalance With the REBALANCE hint, typically used before writing files, AQE splits skewed partitions and merges small ones (optimizeSkewsInRebalancePartitions), giving evenly sized output files.

Related runtime optimisations

• Dynamic partition pruning works alongside AQE: when a filtered dimension table is broadcast, its keys prune partitions of the fact table at runtime.
• Runtime Bloom filters (spark.sql.optimizer.runtime.bloomFilter.enabled, on by default) filter the larger side of a join before the shuffle.

Customisation spark.sql.adaptive.optimizer.excludedRules disables individual AQE rules, and spark.sql.adaptive.customCostEvaluatorClass plugs in your own cost evaluator.

Checking what happened In the SQL tab of the Spark UI, the final plan shows AQEShuffleRead nodes with "coalesced" or "skewed" labels, and joins that changed strategy.

Yes. AQE is on by default because it helps most batch queries, but it only acts at shuffle and broadcast boundaries, only fixes certain problems, and in some cases its choices cost time.

Where AQE can't help

• No exchange, no adaptation: a query that reads, filters and writes without a shuffle has nothing for AQE to re-optimise. The first stage, including how files are split into partitions, is planned statically.
• Streaming queries: AQE doesn't apply to Structured Streaming.
• Skew outside sort-merge joins: skew join handling covers sort-merge joins (and rebalancing). Skewed aggregations, skewed window functions and skew in broadcast or shuffled hash joins still need manual fixes such as salting.
• Some join types: skew can only be split on the side that the join type allows, for example not the right side of a left outer join.
• Bad upstream plans: AQE adjusts the plan after stages run; it won't rescue a cartesian product or a UDF that blocks pushdown.
• RDD code: AQE works on DataFrames and SQL only.

Where it can hurt or surprise

• Late broadcast conversion: switching to a broadcast join after both sides have already been shuffled is less efficient than planning a broadcast from the start. Broadcast hints or good statistics avoid the wasted shuffle.
• Too much coalescing or too little: by default (parallelismFirst true) coalescing ignores the 64 MB target and keeps parallelism high, which can create many small tasks on a busy shared cluster. Setting it to false packs partitions to the target, which can reduce parallelism for CPU-heavy work on small data.
• Output layout changes: coalescing changes the number of output partitions, so the number and size of written files can vary from run to run.
• Forced skew handling: forceOptimizeSkewedJoin may add an extra shuffle.
• Re-planning overhead: each stage boundary adds a little planning time, which matters only for very short queries.
• Broadcast memory: a converted broadcast join can put pressure on driver and executor memory if the threshold has been raised too high.

How to manage it

• Keep AQE on, and tune rather than disable it: initialPartitionNum, advisoryPartitionSizeInBytes, parallelismFirst, and the skew thresholds.
• Disable individual rules with spark.sql.adaptive.optimizer.excludedRules if one consistently hurts a workload.
• Use hints where you know better (BROADCAST, REPARTITION, REBALANCE).
• Keep statistics current so the initial plan is good, and compare the initial and final plans in the Spark UI.
• Handle known skew in aggregations yourself with salting, or with a two-stage aggregation.

Bottom line AQE's downsides are small and mostly about predictability; turning it off globally is rarely right. The real limitation is that it only fixes partition sizing, join strategy and join skew at shuffle boundaries, so plan design still matters.

AQE doesn't handle failures itself. Fault tolerance in Spark comes from the scheduler, task retries and lineage, and those work the same way whether AQE is on or off. What AQE adds is that the plan of the remaining stages is decided only after earlier stages finish, and recovery fits around that.

What happens when something fails

• A task fails: the task scheduler retries it, on another executor if possible, up to spark.task.maxFailures (4) times. If it still fails, the stage fails and the query fails.
• An executor is lost: its running tasks are rescheduled. If shuffle output on that executor is lost, the DAG scheduler re-runs only the map tasks that produced it, using lineage. An external shuffle service or shuffle tracking reduces how often this happens.
• A fetch fails in a later stage: the scheduler re-runs the parent stage that produced the missing shuffle files, then continues.
• Stage retries are limited by spark.stage.maxConsecutiveAttempts (4 by default).

How this interacts with AQE

• AQE breaks the query into query stages at shuffle and broadcast exchanges. A completed query stage's output and statistics are kept; when a downstream stage needs recomputed input, the scheduler regenerates the lost partitions rather than re-planning the query.
• Re-optimisation only happens between query stages, based on completed stages, so a task retry inside a stage doesn't change the plan.
• Because AQE plans are built from runtime statistics, re-running the whole query can give a different final plan (for example a different number of coalesced partitions), but the results are the same.
• A converted broadcast join that runs out of memory fails like any broadcast join; the fix is a lower threshold or more memory, not a retry.

Speculative execution spark.speculation (off by default) launches duplicate copies of very slow tasks and uses whichever finishes first. It helps with slow nodes, not with data skew, and it's separate from AQE.

What actually makes jobs resilient

• Idempotent outputs: writes to table formats such as Delta Lake commit atomically, so a failed job leaves no partial data and can be rerun.
• Checkpoints or persisted intermediate results for very long lineages.
• Sizing executors so tasks don't run out of memory, and fixing skew, which together cause most repeated task failures.
• For streaming queries, checkpoint locations let a query restart from the last committed batch.

In an interview Say clearly that AQE is an optimiser, not a recovery mechanism; explain task retries, stage re-execution and lineage; then explain that AQE works with them because it re-optimises only between completed stages.

Use dropDuplicates (alias drop_duplicates) with the columns that define a duplicate. When it matters which row you keep, use a window with row_number and an explicit order instead, because dropDuplicates keeps an arbitrary row.

from pyspark.sql import functions as F, Window

# 1. Any one row per key
df.dropDuplicates(["customer_id"])

# 2. Fully identical rows
df.dropDuplicates()        # or df.distinct()

# 3. The latest row per key: deterministic
w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())
latest = (df.withColumn("rn", F.row_number().over(w))
            .filter("rn = 1")
            .drop("rn"))

# 4. Latest values with an aggregation
latest_agg = (df.groupBy("customer_id")
                .agg(F.max_by("email", "updated_at").alias("email"),
                     F.max("updated_at").alias("updated_at")))

Choosing between them

• dropDuplicates(cols): simplest and fast; good when all duplicates are equivalent.
• row_number over a window: when you need the newest, the highest-priority or the most complete row. Add a tie-breaker column to the ordering so the result is repeatable.
• max_by / min_by with groupBy: compact when you only need a few columns from the chosen row.
• SQL: QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY updated_at DESC) = 1 where supported (Databricks SQL), or a subquery with row_number in open-source Spark.

Normalise before comparing Duplicates often differ only in case or spaces:

df.withColumn("email_key", F.lower(F.trim("email"))).dropDuplicates(["email_key"])

In streaming

• dropDuplicates(["event_id"]) keeps every key ever seen in state, so add a watermark: withWatermark("event_time", "1 hour").dropDuplicates(["event_id", "event_time"]).
• dropDuplicatesWithinWatermark(["event_id"]) (Spark 3.5 and later) removes duplicates that arrive within the watermark delay even if their event times differ.

When writing to tables To avoid inserting duplicates already in the target, deduplicate the incoming batch, then use MERGE with a WHEN NOT MATCHED THEN INSERT clause in Delta Lake or another table format.

Cost All of these shuffle data by the key columns. Deduplicate after filtering, and select only the columns you need first.

Read the file as lines, split each line into words, normalise them, count by word and take the top N. The DataFrame API does this in a few lines and distributes both the splitting and the counting.

from pyspark.sql import functions as F

N = 10

words = (spark.read.text("/data/books/*.txt")                      # one row per line, column "value"
           .select(F.explode(F.split(F.lower("value"), r"[^a-z']+")).alias("word"))
           .filter(F.col("word") != ""))

top_n = (words.groupBy("word").count()
              .orderBy(F.desc("count"), "word")                      # tie-break for stable results
              .limit(N))

top_n.show()

What each step does

• spark.read.text reads large files in parallel, split into partitions (uncompressed or splittable-compressed text; a single gzip file is read by one task).
• lower and a regular expression that splits on anything other than letters and apostrophes handle case and punctuation. Adjust the pattern for other languages, for example \P{L}+ to split on anything that isn't a letter in any alphabet.
• explode turns each line's array of words into rows; empty strings from leading separators are filtered out.
• groupBy("word").count() aggregates partially on each partition before the shuffle, so only one count per word per partition moves over the network.
• orderBy(...).limit(N) is planned as a top-N operation (TakeOrderedAndProject): each partition keeps only its own top N, so Spark doesn't sort the full result.

Removing stop words

from pyspark.ml.feature import StopWordsRemover
stop = set(StopWordsRemover.loadDefaultStopWords("english"))
words = words.filter(~F.col("word").isin(stop))

The RDD version, for comparison:

import re
counts = (spark.sparkContext.textFile("/data/books/*.txt")
            .flatMap(lambda line: re.split(r"[^a-z']+", line.lower()))
            .filter(None)
            .map(lambda w: (w, 1))
            .reduceByKey(lambda a, b: a + b))
counts.takeOrdered(N, key=lambda kv: -kv[1])

The DataFrame version is faster in Python because the work stays in the JVM, and it also runs on Spark Connect.

For very large vocabularies An exact count is usually fine, since the distinct-word table is small compared with the text. If it isn't, an approximate method such as a count-min sketch (the count_min_sketch aggregate function) estimates frequencies with fixed memory.

Try each expected format with try_to_date (or try_to_timestamp) and take the first that succeeds with coalesce. Rows that match no format come out as null, so you can count and inspect them. Since Spark 4, ANSI mode is on by default, so plain to_date throws an error on a string that doesn't match its format; the try_ functions return null instead.

from pyspark.sql import functions as F

formats = ["yyyy-MM-dd", "dd/MM/yyyy", "MMM d, yyyy", "yyyyMMdd"]

parsed = raw.withColumn(
    "order_date",
    F.coalesce(*[F.try_to_date("raw_date", f) for f in formats]),
)

unparsed = parsed.filter(F.col("order_date").isNull() & F.col("raw_date").isNotNull())

With input 2026-10-01, 01/10/2026, Oct 1, 2026 and 20261001, all four parse to the date 2026-10-01; a value such as bad becomes null.

Things to decide

• Order matters for ambiguous formats: 01/10/2026 is 1 October as dd/MM/yyyy but 10 January as MM/dd/yyyy. Put the format your sources really use first, or choose the format per source system rather than guessing per row.
• Pattern letters follow Spark's datetime patterns: yyyy year, MM month number, MMM month abbreviation, dd day, HH 24-hour hour. Patterns are case-sensitive (mm is minutes, not months).
• Language of month names: MMM parses English names by default.
• Timestamps: use try_to_timestamp with patterns such as yyyy-MM-dd HH:mm:ss, and to_date on the result if you only want the date. Timestamps without a time zone are interpreted in the session time zone (spark.sql.session.timeZone).
• Epoch numbers: timestamp_seconds or timestamp_millis for Unix times.

Formatting for output Once you have a proper DATE column, use date_format(col, "yyyy-MM-dd") only when a string is required for output; keep the date type for filtering, partitioning and date arithmetic.

Clean first Trim spaces and remove known suffixes before parsing:

cleaned = F.trim(F.regexp_replace("raw_date", r"(st|nd|rd|th),", ","))

Data quality Count the nulls produced by parsing and keep the unparsed rows in a quarantine table rather than silently losing them, so new formats from upstream are noticed.

First decide what a missing email or phone number means for the business: is the record unusable, usable with a gap, or recoverable from another source? Then make "missing" consistent (nulls, empty strings and placeholders all become null), measure it, and apply one of a few strategies: reject or quarantine, flag, fill from another source, or keep the null and let downstream logic handle it. Inventing values for contact fields is almost never right.

1. Make missing values consistent Empty strings, spaces and placeholders such as N/A should all become null:

from pyspark.sql import functions as F

contact_cols = ["email", "phone_number"]
placeholders = ["", "n/a", "na", "null", "none", "-"]

clean = df.select(
    "*",
    *[F.when(F.lower(F.trim(c)).isin(placeholders), None)
       .otherwise(F.trim(c)).alias(f"{c}_clean") for c in contact_cols],
)

2. Measure

clean.select([F.sum(F.col(f"{c}_clean").isNull().cast("int")).alias(c) for c in contact_cols]).show()

Track these counts over time; a sudden rise usually means an upstream change.

3. Choose a strategy per use case

• Reject or quarantine: if a record can't be used without contact data (for example, sending invoices), write rows missing both fields to a quarantine table with the reason, and keep them out of the main table.

usable = clean.dropna(subset=["email_clean", "phone_number_clean"], how="all")
rejected = clean.filter(F.col("email_clean").isNull() & F.col("phone_number_clean").isNull())

• Flag: keep the row and add columns such as has_email and contactable, so downstream teams can filter.
• Fill from another source: look up the customer in a master or CRM table and use coalesce(email, crm_email), recording where the value came from.
• Keep null: for analytics, null is honest. Use fillna("unknown") only for display or for grouping in reports, never in the stored data, because a placeholder looks like a real value.

4. Validate the values that are present Missing isn't the only problem: check formats (email pattern, phone numbers normalised to E.164 with a library such as phonenumbers in a pandas UDF) and treat invalid values like missing ones, with the reason recorded.

5. Enforce it in the pipeline

• Declare expectations: Delta Lake CHECK constraints or NOT NULL, data quality rules in Lakeflow declarative pipelines (expectations that drop, warn or fail), or a testing library such as Great Expectations.
• Fix the source when possible: making a field required in the form or application prevents the problem instead of cleaning it every day.

Privacy Emails and phone numbers are personal data. Mask them in logs and quarantine tables where possible, and limit who can read them.

Read the sensor stream, parse it with an explicit schema, add a watermark on the event-time column, group by a 1-minute window (and the sensor id), and compute avg("temperature"). Write the result with a checkpoint, using append mode to emit each minute once it is final, or update mode for results that update as data arrives.

from pyspark.sql import functions as F

schema = "sensor_id STRING, temperature DOUBLE, event_time TIMESTAMP"

readings = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "sensor-readings")
    .option("startingOffsets", "latest")
    .load()
    .select(F.from_json(F.col("value").cast("string"), schema).alias("r"))
    .select("r.*")
    .filter(F.col("temperature").isNotNull()))

per_minute = (readings
    .withWatermark("event_time", "2 minutes")
    .groupBy(F.window("event_time", "1 minute"), "sensor_id")
    .agg(F.avg("temperature").alias("avg_temperature"),
         F.count("*").alias("readings"))
    .select("sensor_id",
            F.col("window.start").alias("minute_start"),
            F.col("window.end").alias("minute_end"),
            "avg_temperature", "readings"))

query = (per_minute.writeStream
    .outputMode("append")
    .option("checkpointLocation", "/chk/sensor_avg_1min")
    .trigger(processingTime="30 seconds")
    .toTable("iot.sensor_avg_1min"))

Why each piece is there

• Event time: windows are based on when the reading was taken (event_time), not when it arrived, so delayed messages land in the right minute.
• Watermark of 2 minutes: readings up to 2 minutes late are still counted; after that, each minute's state is finalised and removed, so memory stays bounded. Choose the delay from how late your sensors' data really arrives.
• Append mode: each window's average is written once, after the watermark passes the end of the window, so the output table only ever receives final values. That adds latency (window length plus the delay). Use update mode with a sink that supports upserts (for example foreachBatch with a Delta MERGE) if a dashboard needs provisional values sooner.
• Checkpoint: stores Kafka offsets and window state, so the query resumes after a restart without double-counting.

Variations

• Sliding windows: F.window("event_time", "5 minutes", "1 minute") gives a 5-minute average every minute.
• Session windows: F.session_window("event_time", "10 minutes") groups bursts of activity.
• Testing without Kafka: the rate source or a folder of JSON files (spark.readStream.schema(schema).json(path)) works the same way.

Production concerns

• Bad messages: from_json returns null for unparseable values; filter them out and send them to a quarantine table instead of dropping them silently.
• Units and calibration: normalise temperatures (Celsius versus Fahrenheit) before averaging.
• Large numbers of sensors: use the RocksDB state store and watch state size in the streaming query's progress.
• Alerts: a second query, or foreachBatch, can flag minutes where the average crosses a threshold.

Normalise the value (trim spaces, lower-case), then test it against a practical regular expression with rlike, and flag or quarantine rows that fail. A regular expression checks the format only: it can't tell you whether the mailbox exists, so treat it as a first filter.

from pyspark.sql import functions as F

EMAIL_RE = r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}$"

checked = (df
    .withColumn("email_clean", F.lower(F.trim("email")))
    .withColumn("email_valid", F.col("email_clean").rlike(EMAIL_RE)))

valid = checked.filter("email_valid")
invalid = checked.filter(~F.col("email_valid") | F.col("email_valid").isNull())

On sample data, " Ann@Example.com " becomes ann@example.com and passes; bo@x (no domain ending) and cy@@x.com fail; di@mail.co.uk passes; a null email gives a null result, which is why the invalid filter checks for null too.

About the pattern

• It requires one @, a local part of letters, digits and ._%+-, a domain of labels separated by dots, and a final part of at least 2 letters.
• The full standard for email addresses (RFC 5322) allows things like quoted local parts that almost no real system uses; a stricter regular expression that follows the standard exactly is long and rarely helpful.
• Internationalised addresses (non-ASCII letters) fail this pattern; widen it if your users have them.
• Lower-casing is safe for the domain; the local part is technically case-sensitive, but practically every provider treats it as case-insensitive.

Going further

• Extra checks with column expressions: length under 254 characters, no consecutive dots (~F.col("email_clean").contains("..")), domain in a list of known disposable-email domains.
• Library-based validation in a pandas UDF (for example email-validator), which can also check the domain's DNS records, but that is slow and needs network access from executors; do it outside the main pipeline if at all.
• Real verification only comes from sending a confirmation email; that belongs in the application, not the data pipeline.

Handling invalid values Don't silently drop them: keep the original value, a validity flag and a reason, write failures to a quarantine table, and count them over time. On Databricks, an expectation in a declarative pipeline (expect_or_drop with an RLIKE condition) or a Delta CHECK constraint enforce the rule in the table itself.

Neither is generally "better"; they do different jobs. wholeTextFiles reads each file as a single record, a (path, whole content) pair, which suits many small files that must each be processed as one unit, such as small JSON or XML documents. CombineFileInputFormat is a Hadoop input format that packs many small files into fewer input splits while still reading them record by record. In current Spark, you rarely need either directly, because the DataFrame readers already combine small files into partitions.

wholeTextFiles

rdd = spark.sparkContext.wholeTextFiles("/data/small_docs/", minPartitions=64)
# RDD of (path, content)

• Good when each file is one logical record and the file name matters.
• Spark already groups many files into each partition (its input format for this extends CombineFileInputFormat), so it doesn't create one task per file.
• Each file must fit in memory, and it's an RDD API: not available on Spark Connect or serverless compute.
• The DataFrame equivalents are spark.read.text(path, wholetext=True) (one row per file) and the binaryFile format, which give file paths and contents as columns.

CombineFileInputFormat

• A Hadoop API (with subclasses such as CombineTextInputFormat) that combines blocks of many files into splits of a target size, so one task reads many small files line by line.
• Used from Spark through newAPIHadoopFile with configuration such as the maximum split size. It means writing low-level Hadoop configuration and usually Scala or Java classes.
• Useful in old Hadoop-format pipelines; little reason to use it in new code.

What Spark does for you now For Parquet, ORC, JSON, CSV and text, DataFrame readers pack small files into partitions up to spark.sql.files.maxPartitionBytes (128 MB), counting spark.sql.files.openCostInBytes (4 MB) per file as the cost of opening it. So 100,000 files of 10 KB don't become 100,000 tasks.

The real fix Many small files are slow mainly because of listing and opening costs on object storage. The lasting solution is to compact them: write fewer, larger files (aim for roughly 128 MB to 1 GB), or store data in a table format whose compaction (OPTIMIZE in Delta Lake) rewrites small files into large ones. See the next question for strategies.

Many small files slow Spark down because every file has to be listed, opened and its metadata read, often over object storage with high latency, and because they can create too many small tasks. The fixes are, in order of effect: compact the files, stop creating them in the first place, and tune how Spark reads them in the meantime.

1. Compact existing files

• Table formats: Delta Lake's OPTIMIZE rewrites small files into large ones without changing the data; Iceberg has the rewrite_data_files procedure. Run them on a schedule, or let the platform do it (predictive optimisation and auto compaction on Databricks).
• Plain files: a one-off job reads a folder and writes it back with fewer partitions:

(spark.read.parquet("/raw/events/date=2026-10-15")
    .repartition(16)
    .write.mode("overwrite").parquet("/compacted/events/date=2026-10-15"))

2. Stop creating small files

• Reduce output partitions before writing: coalesce(n), repartition(n) or the REBALANCE hint, which lets AQE produce evenly sized output.
• When writing partitioned tables, repartition by the partition column first, so each task writes one file per partition instead of one per task per partition.
• Don't partition by high-cardinality columns (user id, timestamp); prefer date-level partitions, or clustering (liquid clustering on Databricks) instead of partitions.
• In streaming, use longer trigger intervals or availableNow runs, and enable optimised writes and auto compaction where available.
• spark.sql.files.maxRecordsPerFile only splits large files; it doesn't merge small ones.

3. Tune reading while the files exist

• Spark packs small files into one partition up to spark.sql.files.maxPartitionBytes (128 MB), treating each file as costing spark.sql.files.openCostInBytes (4 MB). Lowering the open cost packs more files per task when files are tiny.
• Read with an explicit schema so Spark doesn't open files to infer it; for Parquet, avoid mergeSchema unless needed.
• Listing: with many directories, Spark lists them in parallel (spark.sql.sources.parallelPartitionDiscovery.threshold, 32 paths by default). Use partition filters so only needed directories are listed, and pathGlobFilter to skip unrelated files.
• Table formats avoid listing altogether: the transaction log or metadata already holds the file list and statistics.

4. Use incremental ingestion for arriving files Instead of re-listing a growing landing folder, use a streaming file source or Auto Loader on Databricks, which tracks which files it has already processed.

How to see the problem In the Spark UI, look at the number of files read, scan time and task count for the scan stage. For a table, DESCRIBE DETAIL (Delta) shows the number of files and total size; an average file size of a few megabytes or less is a sign to compact.

Neither side fits in a broadcast, so the join will be a shuffle join (normally sort-merge). The work is to shuffle as little data as possible, size the shuffle partitions for about 375 GB, deal with skew, and, if the join runs regularly, store the data so the shuffle can be avoided altogether.

1. Shrink both sides before the join

• Select only the columns needed; columnar formats then read much less than 125 GB and 250 GB.
• Filter as early as possible, including partition filters. If one side is filtered by a small dimension table, dynamic partition pruning can skip partitions of the other at runtime.
• Pre-aggregate if the result only needs totals per key.
• Check for duplicate keys that would multiply rows.

2. Size the shuffle With the default 200 shuffle partitions, each partition would be close to 2 GB, causing spills and memory errors. Aim for about 100–200 MB per partition:

spark.conf.set("spark.sql.adaptive.enabled", "true")                              # default
spark.conf.set("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "2000")  # about 375 GB / 200 MB
spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")

AQE then merges partitions that turn out smaller than the target.

3. Handle skew

• AQE's skew join handling (on by default) splits partitions more than 5 times the median and larger than 256 MB.
• Check the top keys; null or placeholder keys (0, "unknown") are common culprits: filter them and handle them separately.
• For extreme hot keys, salt them: add a random number from 0 to n−1 to the key on the larger side and replicate the matching rows n times on the smaller side.

4. Choose the right strategy

from pyspark.sql import functions as F

a = spark.table("sales.events").select("customer_id", "event_ts", "amount").filter("event_ts >= '2026-01-01'")
b = spark.table("crm.interactions").select("customer_id", "channel", "score")

joined = a.join(b, "customer_id")      # sort-merge join, tuned by AQE

• If after filtering one side becomes small (tens to a few hundred MB), broadcast it with F.broadcast(...), as long as executors have the memory.
• SHUFFLE_HASH can beat sort-merge when one side's partitions comfortably fit in memory, since it avoids sorting.

5. For repeated joins: avoid the shuffle

• Bucketing: write both tables bucketed on customer_id with the same number of buckets; Spark can then join them without shuffling either side.
• Storage-partitioned joins with table formats such as Iceberg do the same with partition transforms.
• On Databricks, liquid clustering on the join key improves data skipping and file layout.

6. Cluster resources The shuffle writes several hundred GB to local disk, so use workers with fast local SSDs and enough memory per core (memory-optimised instances help). Enough total cores to run the 2,000 tasks in a few waves keeps the run time down.

7. Verify In the Spark UI, check the join type in the SQL plan, shuffle read and write sizes, spill, and the spread of task durations. Compare before and after each change.

A task that fails again and again on the same input won't be fixed by retries. Find the exact file or records that cause it, decide whether the data can be skipped, repaired or must stop the pipeline, and then process the rest while quarantining the bad part, without silently losing data.

1. Identify the cause

• Read the task's error in the Spark UI (failed stage → failed task → error) or in the executor logs. Read errors such as FAILED_READ_FILE name the file.
• Check whether it's really corruption: a corrupt file, malformed rows, a schema mismatch (a column type changed), or a value that breaks a cast or a UDF. With ANSI mode on by default since Spark 4, an invalid cast fails the task instead of returning null.
• Find which records are involved: read the suspect file on its own, with mode set to PERMISSIVE and a corrupt-record column for CSV or JSON, and use _metadata.file_path to tie rows to files.

2. Skip or quarantine, depending on the problem

Whole files that can't be read (truncated Parquet, bad compression):

df = spark.read.option("ignoreCorruptFiles", "true").parquet("/data/events/")

Spark logs a warning and skips them. Record which files were skipped, and alert on it.

Malformed rows in CSV or JSON:

df = (spark.read
        .schema("id INT, name STRING, amount DOUBLE, _corrupt_record STRING")
        .option("mode", "PERMISSIVE")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .csv("/data/orders/"))

Then split good and bad rows, and write the bad ones with their file path to a quarantine table.

Values that break transformations: replace cast with try_cast, and the to_date family with try_to_date / try_to_timestamp, which return null; guard UDFs against unexpected input; then count the nulls produced.

On Databricks: badRecordsPath writes bad records and unreadable files to a location with the reason, and Auto Loader's rescued-data column captures values that don't fit the schema.

3. Isolate the file if you can't skip automatically Move the bad file to a quarantine folder, process the rest, and reprocess it after repair. If the source is a table format, restore or re-ingest the affected version (Delta's time travel and RESTORE help when a bad write caused it).

4. Fix upstream Report the issue to the producer, add validation where the data is created, and add checks at ingestion (schema enforcement, expectations) so corrupt data is stopped at the edge rather than deep in a pipeline.

5. Don't just raise retries spark.task.maxFailures (4 by default) exists for transient problems such as a lost executor or a network error. Raising it, or turning on speculation, only repeats the same failure on bad data.

6. Decide the policy Agree with the data's owners when skipping is acceptable: for some data (finance, compliance) a single bad record should stop the pipeline, which is what FAILFAST is for; for others (clickstream), quarantining a small fraction is fine as long as it's measured.

Work from the hardware down: leave each node some room for the operating system, size executors at about 5 cores each, divide the node's memory between them with room for overhead, then set partition counts from the data size so every core stays busy without spilling. For 40 nodes of 16 cores and 256 GB, that gives 3 executors per node, about 120 executors with 5 cores and about 72 GB of heap each, and input partitions of around 128 MB.

1. Capacity per node

• Reserve 1 core and about 16 GB for the OS, node manager and other services: 15 cores and 240 GB usable.
• Total usable: 600 cores and about 9.6 TB of memory. So 25 TB cannot sit in memory at once; the job must stream through stages, and caching should be selective.

2. Executor size

• 5 cores per executor: enough parallelism per JVM without the I/O and garbage collection problems of very large executors.
• 3 executors per node (15 cores), so 120 executors; on YARN, leave room for the application master, giving 119.
• Memory per executor container: 240 GB / 3 = 80 GB. Split into about 72 GB of heap (spark.executor.memory) and 8 GB of overhead (spark.executor.memoryOverhead; the default is 10% of executor memory, and PySpark jobs with Python UDFs need more). That is about 14 GB per core.
• If garbage collection pauses are long with 72 GB heaps, switch to more, smaller executors (for example 5 per node with 3 cores and about 40 GB each).

spark-submit \
  --num-executors 119 \
  --executor-cores 5 \
  --executor-memory 72g \
  --conf spark.executor.memoryOverhead=8g \
  --driver-memory 16g \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.files.maxPartitionBytes=256m \
  --conf spark.sql.adaptive.coalescePartitions.initialPartitionNum=20000 \
  --conf spark.sql.adaptive.coalescePartitions.parallelismFirst=false \
  --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256m \
  job.py

3. Partitions

• Input: with the default 128 MB maxPartitionBytes, 25 TB is about 205,000 input partitions, or about 340 waves of 600 tasks. With large columnar files, 256 MB halves the task count and scheduling overhead; the actual bytes read are often far less than 25 TB thanks to column pruning and filters.
• Shuffles: size them from the shuffle data, not the input. If a stage shuffles, say, 5 TB, partitions of about 256 MB need about 20,000 partitions. Set initialPartitionNum high and let AQE coalesce down.
• Keep partitions a multiple of the core count where possible, so the last wave isn't mostly idle.

4. Other factors

• Disk: shuffles on 25 TB jobs can write many terabytes; nodes need fast local SSDs with enough space for shuffle files and spill.
• Network: shuffle-heavy stages are limited by network bandwidth; reduce shuffled data (filter, project, pre-aggregate, broadcast small tables).
• Skew: AQE handles skewed sort-merge joins; check other stages for straggler tasks.
• Storage: read columnar formats with partition and file pruning, write output in files of roughly 128 MB to 1 GB.
• Dynamic allocation: enable it (with shuffle tracking or an external shuffle service) if the cluster is shared or the job's stages vary a lot in size.
• Driver: 16 GB is enough unless the job collects results or plans very large numbers of files.
• Fault tolerance: long jobs will see failures; write output atomically (Delta or Iceberg tables) and break the job into restartable steps, such as one per day of data.

5. Validate Run on a representative slice first (for example 1 TB), check the Spark UI for spill, garbage collection time, skew and idle cores, then scale. Size from what the stages actually do, not from the input total alone.

Compare a slow run with a good run before changing anything. A job slows down for one of a few reasons: there is more input, the data has become uneven (skew), the query plan changed, tasks are spilling to disk, or the cluster got fewer resources. The Spark UI shows which one, stage by stage.

1. Find the stage that grew

Open both runs in the Spark UI (or the history server) and line up their stages. Usually one stage accounts for nearly all the extra time. Everything that follows is about that stage.

2. Read the task summary for that stage

The stage page shows task metrics as a distribution. Compare the median task with the slowest:

3. Check the input

Did the row count or the number of files jump? A source that now delivers 50,000 small files instead of 500 large ones slows the read even when the bytes are the same. A full reload upstream, or a broken filter that stopped pruning partitions, shows up as a much larger "input size" on the read stage.

4. Compare the query plans

In the SQL tab, open the plan of both runs. Look for:

• A join that was a broadcast hash join and is now a sort-merge join. The small side may have grown past the broadcast threshold (10 MB by default for spark.sql.autoBroadcastJoinThreshold).
• A filter that no longer reaches the scan, so a whole table is read.
• Row counts per operator: a join whose output is far larger than its inputs points at duplicate keys.

5. Check the cluster

Fewer executors than usual (a capped autoscaling pool, spot instances lost), a different instance type, or another job sharing the cluster can all make an unchanged job slower.

Fixes, matched to the cause

• Skew: make sure Adaptive Query Execution's skew join handling is on (AQE has been on by default since Spark 3.2.0). It splits a partition that is both 5 times the median and larger than 256 MB, by default. For skew it does not catch, such as in aggregations, salt the hot key or handle it separately.
• Spill: more, smaller partitions, or more memory per core.
• Plan change: fix the statistics or filter that caused it. Use a join hint only when you know the small side stays small.
• Small files: compact the input, or fix the writer upstream.

Prevent the next one

Record each run's duration, input rows and bytes, and output rows. Alert when duration or input moves far from the recent normal. Most slowdowns are visible in those numbers days before anyone complains.

What interviewers want: a method, not a list of settings. Compare against a good run, find the stage, read the task distribution, then fix the cause you found. "I would increase the cluster size" without a diagnosis is the answer they are screening out.

First work out which memory ran out, because there are two kinds. If the JVM heap ran out, the executor log shows a Java OutOfMemoryError. If the container was killed by YARN or Kubernetes for going over its limit, the memory that grew was outside the heap: usually Python worker processes, or native buffers. The fixes are different, so raising spark.executor.memory for every failure often does nothing.

How an executor's memory is split

The cluster manager gives each executor a container sized as the sum of:

• spark.executor.memory: the JVM heap, where Spark's execution and storage memory live.
• spark.executor.memoryOverhead: memory outside the heap. By default 10% of executor memory (40% for non-JVM jobs on Kubernetes).
• spark.executor.pyspark.memory, if set, and off-heap memory, if enabled.

If spark.executor.pyspark.memory is not set, the Python workers share the overhead allowance and Spark does not limit how much they use. A PySpark job with Python UDFs or pandas code is the usual way a container gets killed while the heap looks fine.

Diagnosis

1. Read the executor's last log lines. "Container killed ... exceeding memory limits" means the container limit. java.lang.OutOfMemoryError: Java heap space means the heap.
2. Find the stage that was running. Is it a wide stage (join, groupBy, window) or a stage with a Python UDF or applyInPandas?
3. Look at that stage's task distribution. One task with far more shuffle read than the rest means skew: a single partition is too big, and no setting fixes that alone.

Fixes for the container limit (outside the heap)

• Raise spark.executor.memoryOverhead, or set spark.executor.pyspark.memory so Python gets its own budget.
• Cut what Python holds: process smaller batches, avoid building large Python lists, and replace Python UDFs with built-in functions where you can.
• With applyInPandas or another grouped map, all rows of a group are loaded into memory at once. A large or skewed group needs splitting.

Fixes for the heap

• Raise parallelism so each task handles less data: more shuffle partitions, or let AQE size them. It is the simplest fix when one task's working set is too large.
• Fix skew: AQE skew join handling, salting a hot key, or treating the hot key separately.
• Fewer cores per executor. Concurrent tasks share the heap, so 8 cores with 16 GB gives each task about half as much as 4 cores with 16 GB.
• Stop caching data you do not reuse. Cached blocks compete with execution memory.

What not to do

Doubling memory until the error stops hides skew and makes every run cost more. It also tends to fail again when the data grows.

What interviewers want: the difference between heap and overhead, that PySpark's Python workers live outside the heap, and checking for skew before adding memory.

A stream falls behind when each micro-batch takes longer to process than the time it took for that much data to arrive. Compare the input rate with the processing rate in the query's progress reports, then find out what makes batches slow. The usual causes are state that keeps growing, a slow sink, too little parallelism, or a sudden surge of input.

1. Confirm it with the progress metrics

Each trigger reports its progress (query.lastProgress, or a StreamingQueryListener that sends it to your monitoring):

p = query.lastProgress
p["inputRowsPerSecond"], p["processedRowsPerSecond"]
p["durationMs"]          # where the batch time went
p["stateOperators"]      # numRowsTotal per stateful operator

If processedRowsPerSecond stays below inputRowsPerSecond, the backlog grows. For Kafka, also watch consumer lag on the topic.

2. Find out why batches are slow

3. Fixes

• Bound the state. Add withWatermark to aggregations, deduplication and stream-stream joins, so Spark can drop state for windows that can no longer change. Without it, state grows for as long as the query runs.
• Move large state off the heap. With millions of keys, switch to the RocksDB state store (spark.sql.streaming.stateStore.providerClass set to the RocksDBStateStoreProvider class). The default store keeps state in executor memory, where millions of keys cause long garbage collection pauses.
• Fix the sink. Write in bulk inside foreachBatch rather than row by row, and keep the target table free of small files. If the sink is a MERGE, check how much of the table each batch rewrites.
• Cap the batch size. maxOffsetsPerTrigger for Kafka or maxFilesPerTrigger for files keeps catch-up batches a size the cluster can finish. Changing these rate limits is allowed between restarts.
• Add parallelism. More Kafka partitions, more cores, or a repartition on a better-spread key.

4. Check the trigger

If a batch runs past the trigger interval, Spark starts the next one at once. That keeps the job busy but does not make it faster. Falling behind is a throughput problem, so a shorter interval will not help.

Prevent it

Alert on the gap between input and processed rates, on Kafka lag, and on state size, not only on query failures. A stream that is losing ground keeps running without errors until the backlog is hours long.

What interviewers want: reading the progress metrics, knowing that unbounded state makes a stream slower every hour, and the watermark and rate-limit fixes.

The checkpoint stores the query's progress (offsets) and its state, and both assume the query keeps the same shape. Some code changes break that assumption: a different source, a different file sink path, or any change to a stateful operation (the keys of an aggregation, the columns of a dedup, a stream-stream join). Spark does not support those changes on an existing checkpoint, and the restart fails or behaves unpredictably. You either undo the change, or start a new checkpoint and decide how to rebuild what the old one held.

What is safe and what is not

What to do now

1. Can you roll the change back? If the change is not urgent, revert it, restart from the existing checkpoint, and plan the change properly.
2. If the change must go in, start a new checkpoint. Point the query at a new checkpoint location and decide where it should start reading:
   • Stateless query: start from the offsets the old query had reached (for Kafka, set startingOffsets from the last committed offsets), so nothing is skipped or read twice.
   • Stateful query: the old state is gone. Rebuild it by reprocessing from a point early enough to cover the state you need, for example the start of the open aggregation windows, and make the sink tolerate the replay.
3. Make the sink tolerate replays. foreachBatch gives at-least-once delivery by default. Use the batchId to skip a batch already written, or write with a MERGE on a key, so reprocessing does not create duplicates.

How to avoid it next time

• Treat a stateful query's keys and state schema as a contract. Changing them is a migration, planned like a table schema change.
• Version checkpoint paths with the query (/checkpoints/orders_agg/v3), so a new version gets a clean location by design.
• Keep sinks idempotent from the start, because every recovery plan depends on replaying data safely.
• If state must evolve, store it as bytes in a format that supports schema evolution, such as Avro.

Never do this

Do not delete the checkpoint folder of the running query and restart in place. The query would then start from its startingOffsets setting, which by default for Kafka is the latest offset, and silently skip everything in between.

What interviewers want: that the checkpoint holds offsets and state tied to the query's shape, which changes are not allowed, and a recovery plan that neither loses nor duplicates data.

Make every write idempotent: running the job twice for the same input must leave the table exactly as running it once. The job was written as a plain append, so the rerun added again the rows the failed run had already committed. Fix the data first, then change how the job writes: replace a defined slice of the table, merge on a key, or tag the write so the table rejects a repeat.

1. Clean up what is there

• Find the duplicates by business key and load date, and confirm the count matches what the failed run wrote.
• On Delta Lake, you can see the two commits in the table history (DESCRIBE HISTORY). Remove the extra rows with a DELETE or a rewrite of the affected slice, inside one transaction.
• Tell the owners of downstream tables: anything that read the table in between read duplicates too.

2. Pick an idempotent write pattern

Replace a slice

(daily.write.format("delta")
   .mode("overwrite")
   .option("replaceWhere", f"event_date = '{run_date}'")
   .save(path))

The rows matching the condition are replaced atomically, and the rest of the table is untouched. A rerun for the same date produces the same result. For a plain Parquet table, the nearest equivalent is dynamic partition overwrite (partitionOverwriteMode set to dynamic). Without it, an overwrite in Spark's default static mode deletes every partition matching the write's specification, which with no specification means the whole table. On Delta, dynamic partition overwrite is now marked legacy, so prefer replaceWhere.

Merge on a key

Use when the input can repeat rows or update them. Deduplicate the batch on the key first: a MERGE can fail when several source rows match the same target row, because it is unclear which one should update it.

Tag the transaction

(batch.write.format("delta").mode("append")
   .option("txnAppId", "orders_loader")
   .option("txnVersion", batch_number)
   .save(path))

Delta records the pair. A later write with the same app id and a version that is not higher than the last one recorded is ignored, so a retried append is not applied twice. The version must be the same on every retry of the same data and higher than the last one written: a batch number works, and so does the time of the first attempt if the job saves it and reuses it on retry. A version taken fresh from the clock on each attempt defeats the check.

3. Make reruns part of the design

• Parameterise the job by the slice it processes (run_date), never by "now". A rerun on Tuesday for Monday must produce Monday's data.
• Do the whole write in one transaction. Several separate writes to one table can leave half a run behind.
• Test it: run the job twice for the same date in a test environment and compare row counts and checksums.

What interviewers want: the word idempotent, a concrete pattern for each case (replace a slice, merge, transaction tags), and the habit of parameterising by date so a rerun does exactly the same work.

Something in the job depends on the order of rows or on how data is partitioned, and Spark guarantees neither unless you ask for it. After a shuffle, rows can arrive in a different order on every run. Functions that pick "the first" row, IDs built from partition numbers, a dedup without a rule for which row wins, and the session time zone are the usual causes. The fix is to make every choice explicit.

The usual causes

1. Picking a row without saying which one

df.groupBy("customer_id").agg(F.first("email"))     # which email?
df.dropDuplicates(["customer_id"])                   # which row survives?

Neither says which row it keeps. first() takes whichever row comes first in that run, and it is non-deterministic because row order can change after a shuffle. dropDuplicates documents no rule for which duplicate survives, so do not rely on one. Say which row you want:

w = Window.partitionBy("customer_id").orderBy(F.desc("updated_at"), F.desc("event_id"))
latest = df.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")

Add a unique column as the last tie-breaker. If two rows share the same updated_at, ordering by that alone is still a coin toss.

2. IDs from monotonically_increasing_id()

Its values depend on the partition ID, so the same row gets a different ID when the partitioning differs between runs. Never use it as a key that other tables store. Build keys from the data itself, for example a hash of the business key columns:

df.withColumn("row_key", F.sha2(F.concat_ws("||", "source", "order_id"), 256))

3. Ranking with ties

row_number() over an ordering with ties gives tied rows numbers in an arbitrary order. Use rank() or dense_rank() if ties should share a number, or add a tie-breaker.

4. limit() without orderBy()

Returns some rows, not particular rows. In a pipeline, it is almost always a bug.

5. Time zones

Converting timestamps to dates uses the session time zone, which by default is the machine's local time zone. Two clusters in different regions can put the same event on different days. Set spark.sql.session.timeZone explicitly, normally to UTC, and use full region IDs rather than short names.

6. Inputs that change underneath

"The same input" is not always the same: a source folder that gained a late file, or a table read by name that was updated between runs. On Delta Lake, read a fixed version for a reproducible run (versionAsOf), and record which version each run read.

7. Floating-point sums

Adding the same doubles in a different order can change the last digits. For money, use DecimalType. For other measures, compare with a tolerance.

How to test for it

Run the job twice on the same snapshot with a different number of shuffle partitions, and diff the outputs. Anything order-dependent tends to show up.

What interviewers want: that Spark does not preserve row order across shuffles, the concrete traps (first(), dropDuplicates, monotonically_increasing_id, ties, time zones), and making every choice explicit with an ordering and a tie-breaker.

First establish which memory ran out, from the exact error, then which task, from the Spark UI. java.lang.OutOfMemoryError: Java heap space in an executor log means the heap filled, usually because one task held too much data. "Container killed by YARN for exceeding memory limits", or exit code 137 on Kubernetes, means the container's total went over, usually memory outside the heap such as Python workers. The Stages tab then shows whether every task is big (too few partitions) or one is huge (skew). Only then pick the fix, and raising spark.executor.memory is rarely the first one.

1. Read the error

2. Find the task in the Spark UI

• All tasks of the stage high in peak memory and spill: partitions are too big. Raise spark.sql.shuffle.partitions above the default 200, or repartition() before the heavy step.
• One task 10 times the median: skew on a key. Check the key's distribution; salt it, or rely on adaptive skew-join handling, which is on by default for joins.
• The failure at a join with a small table: a broadcast that is not small any more. Lower spark.sql.autoBroadcastJoinThreshold or drop the hint.

3. Check what changed

A job that worked last month and fails now usually met more data or a new skewed key. Compare input sizes and the top keys between a good and a bad run.

4. Look at the code

# Gathers every value for a key in one task's memory
rdd.groupByKey().mapValues(sum)

# Combines before the shuffle
rdd.reduceByKey(lambda a, b: a + b)
# or, better, a DataFrame aggregation
df.groupBy("key").agg(F.sum("value"))

Python UDFs that build large lists also count against memory outside the heap; spark.executor.pyspark.memory gives them a limit that fails clearly.

5. Then, and only then, size

Raise overhead or executor memory last, keeping memory plus overhead under the machine's container limit, and note that more memory per executor means fewer executors. See Spark Executor Memory.

What interviewers want: the 3 error families and which memory each means, the Spark UI to separate big partitions from skew, a check of what changed in the data, code fixes like reduceByKey and broadcast limits, PySpark memory outside the heap, and memory increases last.