Every question.
Every answer.

First check existing infrastructure, transformations, and latency requirements. A local script or SQL engine may be enough. Spark can still make sense if the pipeline already runs there, but distributed overhead needs a reason.

collect() returns all rows to the driver. Diagnose the driver result size and heap rather than only executor capacity. Keep processing distributed, write results to storage, or inspect a bounded sample.

The scheduling and coordination state: the DAG scheduler that builds stages for each job, the task scheduler that places tasks, the block-manager master's view of cached blocks, the map-output tracker that locates shuffle outputs, plus broadcast and accumulator state and the Spark SQL planning machinery. That is why pulling results back with collect() or toPandas(), or materializing an oversized broadcast, can kill the driver while executors look healthy.

sc.broadcast(value) is an object you create and manage. The driver serializes a read-only value once; executors fetch and cache a copy on first use; tasks read it through .value instead of receiving the object inside every serialized closure. Use it for lookup data consumed by RDD functions or Python UDFs. A broadcast join is a physical plan choice for DataFrames and SQL. Under spark.sql.autoBroadcastJoinThreshold, or with a BROADCAST hint, the planner collects the small side to the driver, builds a hashed relation, and ships it through a BroadcastExchange so the large side joins without a shuffle. pyspark.sql.functions.broadcast only marks a DataFrame for that plan; it returns a DataFrame, not a Broadcast object with .value. Both keep a copy in driver memory and on every executor, so neither suits large data.

Without a broadcast, the dict is pickled into the function itself. PySpark 3.5.7 moves a pickled command larger than the Python broadcast threshold (1 MiB by default) into an internal broadcast, so the job runs, but you cannot reuse or release that copy. An explicit lookup = sc.broadcast(d) serializes the dict once, lets later jobs reuse executor copies, and gives you control of its lifetime. lookup.unpersist() deletes the executor copies; if a later task reads it, Spark re-sends it. lookup.destroy() removes all data and metadata for the variable, and any later use fails. Both are non-blocking by default. The driver still holds the original dict and executors hold copies, so size the lookup against driver and executor memory, and prefer a broadcast join when the lookup is really a table you join with.

Spark merges all three. Values set directly on SparkConf in the application win, then flags passed to spark-submit or spark-shell, then spark-defaults.conf. Builder .config() calls reach that SparkConf when the builder creates a new SparkContext. The trap is getOrCreate() on a session that already exists, as in a notebook or shell. It returns the existing session with a warning that only runtime SQL configurations will take effect, so a builder value such as spark.executor.memory is silently ignored. Some settings must be known before a process starts: in client mode, spark.driver.memory cannot be set from SparkConf in the application because the driver JVM is already running. Put resource settings in spark-submit flags or cluster defaults, and keep query settings in code or session configuration.

spark.conf is the session's runtime configuration. Runtime SQL configurations such as spark.sql.shuffle.partitions are per-session and mutable, so a new value applies to queries planned afterwards. Static SQL configurations such as spark.sql.extensions and spark.sql.warehouse.dir are fixed once the SQL engine starts; setting one raises 'Cannot modify the value of a static config'. Core settings such as spark.executor.memory belong to SparkConf and are read when executors launch. Since Spark 3.0, spark.conf.set or SQL SET on such a key raises CANNOT_MODIFY_CONFIG instead of silently doing nothing. spark.conf.isModifiable(key) tells you in advance. Keys Spark does not recognize are accepted: a custom application key works, but so does a misspelled one such as spark.sql.shufle.partitions, and the real setting stays unchanged. Changing a static setting needs a new application.

Create the client inside the function that runs on the executor, once per partition, and reuse it for every row in that partition. In a run over 1,000 rows in 4 partitions, a client built inside map was constructed 1,000 times; inside mapPartitions it was constructed 4 times. A client created on the driver and captured in the closure fails before any task runs because it cannot be pickled. Two traps remain. The partition iterator is lazy: if the function returns a generator expression and closes the client before returning, the lookups run after close and the task fails. Yield inside try/finally, or materialize the batch before closing. Tasks can also be retried or run speculatively, so writes from foreachPartition must be idempotent. The partition count sets how many clients run concurrently, so size it to what the service can handle.

Not inherently. Both use Spark SQL’s engine. Compare the optimized and physical plans for equivalent expressions, plus runtime metrics and UDF use.

With map(lambda line: line.split()), each input line produces one list element in the output RDD. With flatMap(lambda line: line.split()), the words become individual output elements; a blank line can contribute none. These transformations do not themselves redistribute records between partitions. Choose the output shape needed by the next operation, rather than assuming flatMap is a faster map.

DataFrame.union resolves columns by position, so compatible types can hide a column-order mistake. unionByName aligns column names. Use allowMissingColumns=True only when filling absent fields with null is acceptable, and check schema types explicitly. Neither operation removes duplicate rows automatically; add a separate deduplication step only if the business rule calls for it.

A window can rank rows within each customer while preserving the order columns. Use row_number over a window partitioned by customer and ordered by timestamp descending, then keep row number 1. If timestamps tie, include a stable unique tie-breaker such as order_id to make the selected row well-defined. If the requirement is to return all orders tied for latest, use rank or dense_rank and keep rank 1 instead.

An RDD is a distributed collection of objects with no schema: maximum control, but Catalyst and whole-stage codegen do not help you. A DataFrame adds a schema and expressions, which is what lets Spark SQL optimize the computation and execute it through generated code. A Dataset adds compile-time typing in Scala and Java only — typed lambdas can cost more than equivalent DataFrame expressions, so typing is a developer-experience trade-off, not a performance upgrade. PySpark has DataFrames only.

Keep the transformation a pure function from DataFrame to DataFrame, so a test calls it without reading or writing storage. Start one local SparkSession for the whole test run, for example a session-scoped pytest fixture with master local[1], few shuffle partitions, and the UI disabled, and stop it at teardown; JVM startup is the slow part. Build tiny inputs with an explicit DDL schema that includes nulls and edge cases, call the function, and compare the result with pyspark.testing.assertDataFrameEqual (added in 3.5.0). It checks the schema when the expected value is a DataFrame. Use assertSchemaEqual when the schema itself is the contract. Leave storage, cluster, and performance behavior to integration tests. One environment caveat: in this runtime, PySpark 3.5.7 with NumPy 2.0.2, importing pyspark.testing failed on np.NaN inside pandas-on-Spark, so check the NumPy version in your test environment.

It collects both sides to the driver, so it is meant for small test data. When both arguments are DataFrames it first calls assertSchemaEqual, which compares field count, names, and order, ignores nullability, and compares data types by type name only. Then, with checkRowOrder=False by default, it sorts both row lists by their string form and compares row by row: order is ignored, but duplicate counts still matter. Floats pass within rtol=1e-5 and atol=1e-8. The gaps follow from that design: decimal(10,2) against decimal(38,0) with equal values passes, a map of string to int against a map of string to string passes the schema check, nullable against non-nullable passes, and a list of Rows as the expected value skips the schema check entirely. When precision or nullability is part of the contract, assert it separately.

PERMISSIVE is the default: a malformed record still becomes a row, with the fields that failed to parse set to null. To keep the raw text, your schema must include a nullable string column whose name matches columnNameOfCorruptRecord (default _corrupt_record, from spark.sql.columnNameOfCorruptRecord). Without that column the bad row survives only as nulls; a non-string column raises an AnalysisException. When Spark infers a JSON schema from bad input, it adds the column itself. DROPMALFORMED drops corrupted records. FAILFAST raises MALFORMED_RECORD_IN_PARSING, but only when an action parses the data; building the DataFrame succeeds. Treat the corrupt-record column as an input to a quarantine table, not something to discard. badRecordsPath is a Databricks option, not part of Apache Spark 3.5.7's JSON or CSV sources.

Both come from column pruning in the reader: Spark parses only the fields a query needs, so what counts as malformed depends on the query. Since Spark 2.3, a query on a raw JSON or CSV scan whose referenced columns are only the corrupt-record column is rejected, and 3.5.7 still raises that error. Cache or save the parsed DataFrame first, then filter it. Adding another column is not a real fix: selecting id with _corrupt_record ran, but returned only the syntactically broken line, because the bad amount value was never parsed. count() needs no columns, so fewer problems are detected: CSV DROPMALFORMED returned 2 rows from collect() but 5 from count(), and JSON returned 2 and 3. The CSV guide documents this pruning (spark.sql.csv.parser.columnPruning.enabled). Compute data-quality counts from a cached or persisted parse.

createOrReplaceTempView gives a DataFrame's plan a name in the current SparkSession only. It stores no data, so each query reruns the plan and sees changes in the underlying table, and the view disappears with that session. createOrReplaceGlobalTempView registers the view in the system database global_temp. Every session in the same application can see it, but only by the qualified name global_temp.name, and it lasts until the application stops. saveAsTable writes data files and registers a catalog table that survives the application only if the catalog is persistent. Neither kind of view is cached; use spark.catalog.cacheTable when you want reuse. You also cannot create a persistent view with CREATE VIEW over a temporary view, because the persistent object would reference something that disappears.

Transformation count alone cannot tell you. Ask which actions run, inspect the executed plan, and confirm in the Spark UI. Lazy transformations primarily build a plan.

Lazy planning lets the optimizer see the whole pipeline before anything runs: column pruning, filter pushdown, expression folding, join reordering, and codegen fusion are all rewrites of a plan you have not executed yet. Execute eagerly and each step would materialize data the next step can no longer optimize away. Actions are what trigger the work, and a plan inspection such as explain() shows the result of planning without running it.

Ten waves is only a lower-bound model for uniform tasks with no overhead. Skew, scheduling, I/O, retries, and contention affect actual completion time.

Not by itself. A lazy map does not run until required by an action, and accumulator updates inside transformations may be applied again when tasks or stages are re-executed. Spark documents a once-per-task update guarantee for updates performed inside actions only; that is not a guarantee across repeated application actions. For a business total, represent rejected rows in the data and compute the count as a query result. Use transformation accumulators as diagnostic signals.

A job is a top-level unit of work submitted to the scheduler, typically triggered by an action — though one action can spawn several jobs, because preparation work such as broadcast materialization is itself a job. A stage is a set of tasks separated at shuffle boundaries, per the DAGScheduler source. A task is one partition's worth of work in one stage attempt, and it is retried individually before the stage fails. Jobs and stages appear in the Spark UI; tasks are the rows inside a stage.

Speculation launches a duplicate attempt of a task that is running unusually slowly and keeps whichever finishes first — useful on heterogeneous or flaky clusters where stragglers are caused by the environment, not the data. It is risky when tasks have side effects: both attempts can run, so non-idempotent writes or external counters can double-apply. It also consumes cluster capacity, so on a saturated cluster it mostly duplicates load.

It funnels output through one partition and can reduce upstream parallelism. Check file requirements; prefer a useful number of output partitions. A shuffle before the final write trades extra redistribution for parallel upstream work.

reduceByKey combines values within each partition before merging results for a key. For repeated keys, this can move much less data than grouping all their values and then summing. The reduction function must be associative and commutative; addition fits the intended aggregation, with floating-point rounding caveats. This is an RDD comparison: DataFrame groupBy(...).sum(...) can already use partial aggregation, so do not infer the same penalty from its name.

A narrow transformation computes each output partition from a single input partition, so it pipelines inside the current stage. A wide transformation needs data from many input partitions, which requires a shuffle, and a shuffle boundary breaks the DAG into separate stages — one writes map output files, the next reads them. The DAGScheduler source states it plainly: stages are created by breaking the RDD graph at shuffle boundaries.

repartition runs a full shuffle: it can increase or decrease the partition count, and it redistributes rows evenly unless you partition by columns. coalesce merges existing partitions without a shuffle, so it can only reduce the count, and it can leave the remaining partitions uneven. Use repartition when you need balance or a target count for downstream parallelism; use coalesce when you are trimming file count and can accept inherited skew.

Work backwards from sizes and cores. Spark's file source packs scans up to spark.sql.files.maxPartitionBytes (128 MiB by default), and AQE sizes post-shuffle partitions from an advisory target (64 MiB by default). A practical range is tens to a couple hundred MB per task, enough partitions to keep a wave of tasks busy — commonly 2-3 per core is the documented heuristic — and a count that is a sensible multiple of available cores so the final wave is not mostly idle.

It is the number of partitions used for post-shuffle stages of joins and aggregations, defaulting to a fixed 200 regardless of data size: too many tiny tasks for a small shuffle, too few oversized tasks for a large one. Modern Spark lets AQE coalesce post-shuffle partitions from a deliberately larger initial count using real output sizes, so the setting becomes an upper bound rather than a tuning guess. Check the executed plan to see how many partitions actually ran.

Each map task sorts or hashes its output by the target partitioner and writes it to local disk, keeping an index that lets readers find each block; the default sort-based manager produces one data file and one index file per map task. Reduce tasks then fetch their blocks from every map task's executor over the network, with fetch retries and a fallback after failures. Because map output lives on local disks, losing an executor can lose shuffle files and force upstream stages to rerun.

Throttles around the shuffle fetch client: spark.reducer.maxSizeInFlight caps in-flight fetch memory (48 MiB default), with related limits on concurrent fetch requests and blocks per address. They stop many reducers from simultaneously dragging the cluster's output into memory, which would cause OOMs and fetch failures. On large clusters where fetch wait dominates, these are the knobs to examine — with the understanding that raising them trades network pressure for memory pressure.

Partitioning writes directories keyed by low-cardinality columns, so filters on those columns skip entire directories during listing. Bucketing writes a fixed number of hash-bucketed files inside each partition, which Spark can use to avoid the exchange for joins and aggregations on compatibly bucketed tables — repeated large joins pay the shuffle once at write time instead of every run. Bucketing requires matching bucket counts and stable write behavior; a poor choice of bucket key leaves all its cost and no benefit.

PySpark RDD functions and records live in Python. The driver pickles each function with cloudpickle, together with every variable it captures, and sends it to Python worker processes. Records move between the JVM and those workers as pickled batches; the default context serializer is AutoBatchedSerializer over CloudPickleSerializer. An RDD shuffle such as reduceByKey or partitionBy hashes keys in Python and lets the JVM shuffle key-hash and pickled-byte pairs. A captured object that cannot be pickled, such as a socket, a lock, or a live client, fails on the driver with PicklingError before tasks start. Create such resources inside the function on the executor, typically once per partition. This per-record pickling is also why Python RDD code costs more than DataFrame expressions, which stay in the JVM's binary rows unless a Python UDF pulls them out.

They act at different levels. repartitionByRange(n, col) shuffles rows into n execution partitions that each hold a contiguous key range. Boundaries come from sampling, so they are approximate and can vary between runs. sortWithinPartitions(col) sorts inside each existing partition without a shuffle, while orderBy adds a range exchange to produce a total order. write.partitionBy(col) controls directories: one col=value directory per distinct value, with the column stored in the path rather than in the data files. Combine them by purpose. Use partitionBy for low-cardinality columns that queries filter on, repartition or repartitionByRange on those columns to bound files per directory, and sortWithinPartitions on a high-cardinality filter column so Parquet min/max statistics can skip row groups.

Write tasks determine the file count, not partitionBy. Each task writes a separate file for every day value in its partition, so N tasks whose rows span all days produce up to N files in each directory. With 4 input partitions and 3 days, partitionBy("day") alone wrote 4 files per directory. Adding repartition("day") first hash-partitions on the directory column, so each day's rows meet in one task, and that run wrote 1 file per directory. The cost is a shuffle, plus one task and one file per value: a large or skewed day becomes a huge file and a straggler. Cap file size with the maxRecordsPerFile write option or spark.sql.files.maxRecordsPerFile; with a cap of 50 on the unrepartitioned write, each directory held 8 files. coalesce reduces tasks but does not group rows by day.

Compare the slow task’s input and shuffle size, spill, and execution time. Check hot or null keys and join multiplicity. Consider adaptive skew handling where applicable; do not assume that adding executors splits a single hot partition.

Ordinary equality with null evaluates to unknown, so an inner join on equality does not match null keys. PySpark Column.eqNullSafe, or SQL <=>, matches null with null. Use it only when that matches the business meaning of missing IDs. If the left side has m null keys and the right has n, null-safe equality can produce m × n matches; replacing nulls with a shared sentinel can cause the same multiplication.

A left semi join keeps each left-side row that has a match without returning right-side columns or multiplying it for multiple right matches. A left anti join keeps left rows with no match. Neither operation deduplicates duplicate rows already present on the left. Choose the join condition deliberately: under ordinary equality, null keys do not match null keys.

The hint prioritizes broadcasting the specified relation even when its estimated size exceeds spark.sql.autoBroadcastJoinThreshold. It does not guarantee a hash join: the join type must be supported, and a join without equality keys may use a broadcast nested loop strategy. Check the selected operator and whether the relation can be materialized safely; a hint is not a memory reservation or proof that broadcasting is cheap.

Broadcast hash join: replicate a small side to every executor. Broadcast nested loop join: replicate a side when no equality keys exist. Shuffle hash join: build a per-partition hash map from one shuffled side. Sort-merge join: shuffle and sort both sides, spilling safely — the workhorse for large equi-joins. Cartesian joins cover everything else and are usually a mistake to allow quietly. The strategy appears in the physical plan and in the SQL tab.

Static planning looks at estimated sizes against spark.sql.autoBroadcastJoinThreshold (10 MB default), whether the keys support sorting or hashing, the join type, and any strategy hints. Runtime takes over when adaptive query execution is enabled (it is, since 3.2): after the shuffle map stage completes, AQE can convert a sort-merge join to a broadcast or shuffled hash join based on actual sizes. Always read the chosen operator from the executed plan rather than assuming the threshold decided it.

When the broadcast side is only nominally small. The driver materializes the relation to broadcast it and every executor must hold a copy, so an oversized choice surfaces as driver memory pressure, broadcast timeouts, or executor OOMs rather than as a fast join. Eligibility also depends on the join type and keys — check which operator was actually selected. A hint prioritizes broadcasting; it is not a promise that the data fits.

Sort-merge join sorts both shuffled sides and merges them: it spills to disk and stays safe at scale, which is why it is the default for large equi-joins. Shuffled hash join builds a hash table from one side per partition: less sorting work and often faster, but the build side must fit in execution memory or the task OOMs. AQE converts sort-merge to shuffled hash when every post-shuffle partition is below the local-map threshold — evidence the build side is genuinely small.

Two mechanics. First, unmatched rows from the preserved side must survive into the result, so there is less the optimizer can drop or push down through the plan. Second, null join keys never match under equality — the SQL null semantics apply — so null rows pass through unjoined rather than joining; problems start when you rewrite nulls into a shared sentinel value or use null-safe equality, which can send every such row to one partition (and multiply matches).

Assume a shuffle join, then try to avoid paying for it repeatedly. If the filterable part of the dimension is small enough, filter and aggregate it first, then broadcast. If this join runs every night, bucket both tables on the join key so Spark can skip the exchange. If one key dominates the fact table, handle skew explicitly — AQE skew handling on sort-merge joins, or salting — rather than hoping a bigger cluster fixes one hot partition.

Prefer the cheapest structural fix that addresses the cause: enable or trust AQE skew handling, which splits and replicates skewed sort-merge join partitions from real post-shuffle sizes; salt the hot key when AQE cannot help (for example in a groupBy shuffle); isolate the hot keys into a separate broadcast join or pre-aggregate them; or reduce the incoming data with a filter. Runbook order matters: identify the key distribution first, since each fix assumes something about it.

For the skew-prone key K: on the large side, append a random bucket label to K (roughly N values), producing K#0..K#N-1. On the small side, duplicate each row for K into N copies sharing those labels — typically by exploding an array of range(N). Join on the composite key so each bucket is a normal-sized partition, then aggregate the partial results, dropping the salt. Salting multiplies the small side and adds a final shuffle, so use it when a hot key genuinely dominates and AQE skew handling is not applicable.

No. Identify repeated expensive computation, memory capacity, and reuse lifetime. Caching one-use intermediates can add overhead and pressure. Measure, inspect storage, and unpersist deliberately.

No. RDD.cache() uses MEMORY_ONLY, while DataFrame.cache() uses MEMORY_AND_DISK_DESER in this version. persist(...) lets you choose a storage level explicitly. Cached data is populated as actions compute it; calling cache() alone does not prove the full dataset is stored. Inspect the storage level and materialized partitions, and release cached data with unpersist() when its reuse ends.

Caching reuses computed data while retaining the dependencies needed for recomputation. DataFrame.checkpoint() returns a DataFrame with a truncated logical plan and writes checkpoint files to the directory configured with SparkContext.setCheckpointDir(); use reliable shared storage on a cluster. localCheckpoint() also truncates the plan but stores data through executor caching, so losing that data can prevent recovery. A batch checkpoint does not by itself restart the driver application or supply streaming recovery semantics.

The JVM heap is sized by spark.executor.memory; inside it Spark reserves 300 MiB for system use and splits the rest. spark.memory.fraction (0.6 default) is the unified region where execution memory — shuffles, joins, sorts, aggregations — and storage memory for cached blocks compete; spark.memory.storageFraction (0.5 default) is the share of that region cached blocks can hold without being evicted. Execution can evict storage, never the reverse, which is why caches appear to shrink under heavy compute.

Read where the error was raised. Driver-side OOMs usually trace to collect(), toPandas(), a large broadcast build, or oversized driver-side structures — the failing process is the driver. Executor task OOMs trace to one oversized unit of work: a hot key, an oversized partition, too few partitions, or non-spillable state inside a UDF. Container kills with a healthy-looking heap point at memoryOverhead or the Python worker, not the JVM heap at all.

Non-heap memory charged to the executor container beyond the JVM heap: JVM overheads, native allocations, and, for PySpark, Python worker processes when no separate allocation is configured. Under the default factor (0.10, with a 384 MiB floor — 0.40 for Kubernetes non-JVM workloads) Spark computes it from executor memory; setting it explicitly overrides that. The classic 'container killed by YARN for exceeding memory limits' with a healthy heap is this budget, which is why raising memoryOverhead — not only heap — is the fix.

Spill is Spark trading memory for disk: when the execution memory manager cannot grant more space to a sort or aggregation, the operator writes sorted runs to local disk instead of failing. The stage completes more slowly. The UI reports it directly — shuffle spill (memory) and (disk) on stage pages, spill size and peak memory per operator in the SQL tab. Treat it as a signal when it is large relative to the stage input or concentrated in a few tasks; compare tasks rather than expecting a fixed volume.

spark.serializer (default JavaSerializer) serializes JVM objects that are shuffled, sent over the network, or cached in serialized form. Kryo is faster and more compact, and works best with registered classes. The payoff is clearest for Scala or Java RDDs of custom objects and serialized RDD caches. DataFrame and SQL shuffles avoid that path: ShuffleExchangeExec uses UnsafeRowSerializer over Tungsten's binary rows, and encoders convert typed objects into that format. The setting still applies to DataFrames when Spark serializes whole blocks. A DataFrame persisted at a serialized level, such as MEMORY_ONLY in PySpark or DISK_ONLY, stores its cached batches through spark.serializer. PySpark RDD records are already pickled bytes when the JVM sees them, so Kryo changes little there. Switch serializers when profiling points to serialization cost, not by default.

Start with the Jobs tab to find the longest-running job, open its detail page for the stage breakdown, then open the longest stage and compare per-task duration, input, shuffle, spill, and GC time. Cross to the SQL tab to name the expensive operator before changing code or configuration.

A skewed or oversized partition: one key or partition carries a disproportionate share of rows. Check hot or null keys and join multiplicity, compare the straggler against the median rather than the average, and fix the data layout or eligible skew handling before adding executors.

No. AQE changes eligible parts of a SQL physical plan using runtime statistics, such as combining small post-shuffle partitions or changing join strategy. Dynamic allocation changes how many executors the application requests and releases as workload demand changes. One changes execution planning; the other changes available resources. They can work together, but enabling one does not enable the other. Inspect the executed SQL plan for AQE and executor events for allocation changes.

Why so little parallelism: partition count relative to cores, whether the input is one huge splittable file that was not split, thousands of small files collapsing into few tasks or the opposite, and whether an earlier filter or join reduced the data without reducing the partition count. Five tasks means five units of work no matter how big the cluster is — the fix is repartitioning or fixing the input layout, not adding executors.

QueryExecution runs the phases in order. The parsed, unresolved logical plan is analyzed — columns, tables, and types resolved — as soon as the DataFrame is built. The optimizer applies rule batches such as null-filter inference, constant folding, and predicate pushdown. Physical planning applies strategies in a fixed order and, in Spark 3.5.7, takes the first plan they return; the join strategy comes from rules over size estimates, hints, and join type. Statistics make those estimates cost-based only when they are collected with ANALYZE TABLE and spark.sql.cbo.enabled is true, which it is not by default. Preparation rules then insert exchanges and sorts (EnsureRequirements) and collapse codegen stages. explain("extended") prints the parsed, analyzed, optimized, and prepared physical plans; with AQE on, the physical plan is re-planned at run time.

It fuses a chain of operators that support codegen into a single generated Java function — the source file's own summary: 'compiles a subtree of plans that support codegen together into single Java function'. One loop processes each row through filter, projection, aggregate without per-row virtual dispatch or intermediate materialization. The SQL and stage DAGs show WholeStageCodegen blocks with codegen IDs; operators that cannot be fused, like an exchange, break the chain. Codegen can also disable itself and fall back on very large generated methods.

After a shuffle map stage materializes, AQE re-optimizes and re-plans the rest of the query with its measured output sizes. By default it coalesces small post-shuffle partitions, converts a sort-merge join to a broadcast hash join when a side proves smaller than the adaptive broadcast threshold, and splits skewed join partitions. Converting to a shuffled hash join also needs spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold above its default of 0. AQE has been on by default since Spark 3.2.0. Before an action, explain shows AdaptiveSparkPlan isFinalPlan=false; afterwards, the same DataFrame or the SQL tab shows the final plan, with query stages and AQEShuffleRead coalesced, local, or skewed. A runtime switch comes after the shuffle writes, so it saves sorts and fetches, not the map-side shuffle.

Anything that is not a decision made after a completed stage. AQE cannot undo work already done: a runtime broadcast arrives after both sides' shuffles were written. It does not change scan parallelism, file layout, small files, or unpushable predicates, and a Python UDF stays opaque. A query without an exchange is not wrapped in an adaptive plan at all. Skew splitting covers joins and rebalance partitions, not a hot key in a groupBy or window. Coalescing only merges partitions — it never splits an oversized one — so a task that cannot fit its partition remains a problem. Driver-side collects are out of scope. Read the final plan as evidence of what AQE decided, then fix the remaining cause in layout, expressions, or configuration.

Pushdown depends on expression shape. A filter reaches a Parquet reader only if Spark can translate it: a column compared with a literal, IN with literals, null checks, string prefix, suffix, or contains tests, and AND, OR, or NOT of those. A Python UDF becomes its own operator, so the scan gets PushedFilters: []. Wrapping a data column in a cast pushes only IsNotNull. Non-deterministic parts stay in Spark while deterministic conjuncts still push. An OR mixing a partition column with a data column is neither pruned nor pushed. Partition pruning works differently: Spark evaluates any deterministic partition-only predicate against listed values, so cast(day as string) = '2026-09-02' still read 2 of 6 files locally, while a Python UDF on day read all 6. Check PartitionFilters, PushedFilters, and files read.

Collected statistics are only half of it. By default Spark estimates plan sizes with a size-only visitor, and a filter keeps its child's size estimate. Row counts and column statistics drive estimation only when spark.sql.cbo.enabled is true, and it is false by default in 3.5.7. On a local 1,000-row table filtered to amount > 900, the estimate stayed 5.5 KiB with or without table statistics while CBO was off. With CBO on and ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS amount, it became rowCount=100 and 2000 B. Plain ANALYZE collects size and row count; column statistics need FOR COLUMNS. Statistics are not refreshed automatically by default, so re-run the command after large data changes. Join reordering additionally needs spark.sql.cbo.joinReorder.enabled.

Probably not. With adaptive query execution on, the 3.5.7 default, a query containing an exchange is wrapped in AdaptiveSparkPlan, and codegen stages are collapsed for each query stage as the adaptive plan runs. The pre-execution tree, marked isFinalPlan=false, therefore lists operators without stars, and explain("codegen") reported zero WholeStageCodegen subtrees in a local run. After an action, explain() on the same DataFrame printed the final plan with *(1) around the scan, filter, project, and partial aggregate and *(2) around the final aggregate. With AQE disabled, both stages appeared before execution. Missing stars can also be real: ObjectHashAggregate, BatchEvalPython, or a Project whose expression needs the interpreted fallback all sit outside generated code.

Not in Spark 3.5.7. The planner applies its strategies to the optimized logical plan, and QueryExecution takes the first physical plan returned; the source comment says it uses next() for now and will implement choosing the best plan later. Choices that look cost-based are rules over estimates: JoinSelection checks hints, then whether a side's estimated size fits spark.sql.autoBroadcastJoinThreshold, then the shuffled hash conditions, then sort-merge join for sortable keys. Cost-based optimization in this version improves the logical side — cardinality estimates and join reordering — when statistics are collected and spark.sql.cbo.enabled and spark.sql.cbo.joinReorder.enabled are set. At run time, AQE revisits some decisions with measured shuffle sizes, which is the closest thing to feedback.

No — it had already shuffled it. AQE decides after a shuffle map stage finishes and reports its output size, so both sides' map stages have run and written shuffle files before the switch. In a local run, the initial plan was SortMergeJoin because a filtered Parquet side kept a 13.4 MiB file-size estimate against a 1 MiB threshold; the final plan was BroadcastHashJoin, with AQEShuffleRead local over ShuffleQueryStage 0 and 1. The switch saves the sorts and, through the local shuffle reader, reduce-side network fetches. To skip the large side's shuffle entirely, the static plan must choose broadcast: accurate statistics, a smaller estimate, or a broadcast hint on a side you have confirmed fits in memory.

Prefer the built-in expression when its semantics fit: Spark can execute and optimize that expression without calling your Python function. Arrow improves data transfer and serialization across the Python boundary; it does not translate Python code into a built-in SQL expression. In Spark 3.5.7, an Arrow-enabled ordinary Python UDF still runs its function row by row, while a pandas UDF uses a different, batch-oriented interface. Check null handling and return types, and measure the actual workload.

Three costs. The function runs in a separate Python worker process, so each row's input columns are pickled in the JVM, sent over a socket in batches, and unpickled into Python objects, and results travel back the same way. The function is called once per row, so Python call overhead grows with row count. And it is opaque to Catalyst: ExtractPythonUDFs moves it into a BatchEvalPython operator, which splits whole-stage codegen and prevents pushing any predicate that depends on its result. Predicates that do not touch the UDF can still move below it. On one laptop, a trivial upper-case UDF ran about eight times slower than the built-in; treat that as an illustration, not a rule. Prefer built-ins, then vectorized pandas UDFs.

They change the transfer and the calling convention. Data crosses as Arrow record batches, up to spark.sql.execution.arrow.maxRecordsPerBatch rows (10,000 by default), and the function receives a pandas Series per batch — or a whole group for a Series-to-scalar aggregate — instead of being called per row. That removes per-row overhead only when the body is vectorized; Series.apply inside is still per-row Python. The boundary remains: the plan shows ArrowEvalPython, codegen still splits, and predicates on the result are not pushed. Return types must be declared, and for Series to Series the output length must match the input. A trivial local example ran faster than a plain Python UDF but slower than the built-in; the gain depends on the workload, not a fixed multiplier.

First a built-in function or SQL expression: Catalyst can optimize it, push eligible filters, and generate code. Second, higher-order functions such as transform, filter, exists, and aggregate for array and map columns; they avoid Python entirely, though the observed plan evaluated transform outside whole-stage codegen. Third, a pandas UDF or pandas function API when the logic genuinely needs Python libraries, written with vectorized operations. Last, a row-at-a-time Python UDF. useArrow=True speeds its data transfer but still calls the function per row, so it is not equivalent to a pandas UDF; check its type coercion before enabling it. Whatever you choose, confirm in the plan — BatchEvalPython or ArrowEvalPython marks the Python boundary — and time a representative sample.

Choose by what the function needs to see. mapInPandas receives an iterator of pandas DataFrames per partition, sized by spark.sql.execution.arrow.maxRecordsPerBatch, and may return any number of rows. It needs no shuffle; locally it planned as MapInPandas directly over the scan. Use it for batch-local work such as parsing or scoring. applyInPandas passes one pandas DataFrame per group, so it requires a full shuffle: the plan was Exchange, Sort, then FlatMapGroupsInPandas. Use it only when the logic needs all rows of a key together. Its risk is memory: each whole group is loaded into the Python worker, maxRecordsPerBatch does not split it, and a skewed key can fail the task. If a built-in aggregate or window function expresses the logic, prefer that.

Arrow changes the conversion format, not where the data ends up. With spark.sql.execution.arrow.pyspark.enabled=true, results reach the driver as Arrow record batches and become a pandas DataFrame far more cheaply than row-by-row conversion; a local 500,000-row, two-column example took 0.42 s with Arrow against 1.46 s without, on one laptop. But toPandas() still loads every row into driver memory, and the API notes say to use it only when the result is expected to be small. The experimental spark.sql.execution.arrow.pyspark.selfDestruct.enabled option exists to reduce memory during conversion. Aggregate or sample in Spark first, or write large results to storage. If Arrow conversion fails before computation, Spark falls back to the slower path unless spark.sql.execution.arrow.pyspark.fallback.enabled is false.

It caps how many rows Spark puts into one Arrow record batch, and so how large each pandas Series or DataFrame your function receives. The default is 10,000 rows. Locally, a pandas UDF over one 25,000-row partition received batches of 10,000, 10,000, and 5,000 rows; at 1,000, every batch had 1,000. The cap counts rows, not bytes, so wide rows or large strings need a lower value, and the guide notes that building batches can temporarily raise JVM memory. It does not help grouped APIs: applyInPandas and cogroup load each whole group as one batch. Also check where the pandas objects live — Python worker memory, budgeted by memoryOverhead or spark.executor.pyspark.memory.

More than once. Spark treats a UDF as deterministic unless you call asNondeterministic(), and the API notes warn that optimization may remove duplicate invocations or invoke the function more times than it appears. In a local plan, filtering on a selected UDF column let the optimizer push the filter below the projection by copying the UDF expression, so two BatchEvalPython operators evaluated it. Marked non-deterministic, the filter stayed above a single BatchEvalPython. Task retries and speculative attempts can still repeat calls, so non-determinism is not exactly-once. For side effects, avoid a UDF: call the service in a separate, idempotent step — for example mapInPandas with explicit batching — and join the results back.

Row-based files (Avro, JSON lines, CSV) keep one record's fields together and fit write-once, read-whole-record flows: streaming ingest, Kafka payloads, log shipping. Columnar files (Parquet, ORC) keep one column's values together and fit analytic scans over a few columns of many rows, where they win three separate ways: only projected columns are read, same-type values compress harder, and vectorized readers decode contiguous batches. Pick by read pattern first, write pattern second.

Splittability is a property of the container, not the codec. A raw gzip stream has no internal boundaries a reader can seek to, so the whole file is one unit of work. Parquet compresses each page independently and explicitly separates metadata from data, so any reader first reads the footer and then reads row groups and column chunks selectively — the container stays splittable regardless of which codec is applied inside. The Parquet version also reads only the needed columns and skips text parsing entirely.

It starts at the end: magic bytes and footer length, then the footer — the schema, every column-chunk offset, and per-chunk statistics — and only then reads the chunks and pages the query needs. The footer is written last because offsets and statistics can only be known after all row groups are written, which is exactly why Parquet is write-once: you cannot append a row group in place, and new data becomes new files. That same property is what lets table formats commit atomically by adding files.

Row groups are the file's horizontal slices; each row group stores one column chunk per column; each chunk is divided into pages, the smallest unit of encoding, compression, and statistics. Row-group statistics let a scan drop whole groups; page-level indexes allow finer skipping inside a chunk. The sizes trade off in both directions: small row groups mean more metadata and weaker compression, large ones mean coarser skipping and larger reads.

Min/max statistics can only drop a row group when the predicate range cannot intersect the group's range. If the data was written in arbitrary order, every row group's min/max spans the entire domain — the statistics are correct and useless. The fix happens at write time: sort or cluster the data on the filter column, or make it the partition column so whole directories can be skipped.

Dictionary encoding stores each distinct value once and encodes rows as indexes — excellent for repetitive data, but a near-unique column like UUIDs keeps growing the dictionary until it exceeds its budget, and the writer falls back to plain encoding, storing every full value. Consider whether the raw identifier must be stored for these scans, or whether a hashed or bucketed surrogate is enough for analytics.

Bloom filters: a compact per-chunk structure that can prove a value is absent, so equality predicates like user_id = 'abc' can skip chunks even when the ranges overlap everywhere. They add write cost and file size, so enable them per column where equality filtering dominates. Spark exposes them as an ORC table option; Parquet supports them too.

Vectorized decoding reads columnar batches (4096 rows per batch by default) and is far friendlier to CPU caches than row-at-a-time decoding. Historically nested types forced the row path; Spark 3.3 added nested-column vectorization, enabled by default, on top of the base reader. The config table also shows that record-level filtering only takes effect when the vectorized reader is disabled. It matters because a plan can look columnar while the reader quietly decodes row by row.

Partition pruning happens during directory listing, before any file opens — directory names encode values. Row-group and file skipping happens once the footer is read, using statistics. Column pruning happens from chunk offsets, so unreferenced columns are never read. Predicate pushdown is the mechanism that hands supported filters to the reader so pages can be skipped while decoding. When a scan is slow, name which level failed before changing any configuration.

Add: yes — older files simply read null for the column. Rename or drop: not safely in plain Parquet, because resolution is by name and old files still carry the old name; table formats add ID or name-mapping layers that make those metadata-only operations. Type changes are safe only when widening (int to long, float to double, higher decimal precision). And reading a directory whose files disagree on schema requires mergeSchema, which reads every file footer — off by default for a reason.

A transaction or table layer that defines which files constitute the table at each version. Parquet files are immutable and object stores have no atomic multi-file commit, so a plain directory of files provides no ACID transactions, no safe concurrent writers, and no time travel. Iceberg swaps table metadata files atomically and tracks snapshots through manifest lists; Delta commits ordered log entries that add and remove files. Spark reads each format through its own connector, and their guarantees come from those projects rather than from Spark itself.

Both are columnar containers with statistics, compression, and vectorized readers, and Spark supports each natively. ORC's Spark integration exposes built-in bloom filters and a native implementation that has been the default since 2.3, with zstd and columnar encryption available as table options. Parquet has the broader ecosystem because most engines default to it. For a given lakehouse the decision is usually ecosystem-driven; the fundamentals are shared.

Avro is row-based and built for writing and moving whole records: Kafka messages, CDC events, landing zones. Its distinguishing feature is schema resolution — reader and writer schemas are reconciled at read time using field names and defaults, so producers and consumers can evolve independently. In Spark, from_avro and to_avro decode and encode Kafka payloads, and the module is external: add it with --packages. The common pattern is Avro in the landing layer, Parquet for analytics.

Parquet and ORC are both columnar, with statistics, compression, predicate pushdown, and vectorized readers in Spark 3.5.7. Pushdown is not what separates them; ecosystem usually is. Parquet is the common default for Spark lakehouses, and Delta Lake stores its data as Parquet. ORC is at home where Hive is. The draft's 'better ACID legacy in Hive' needs attribution: ACID is a Hive feature. The ORC project documents Hive transactional tables built from ORC base and delta files, but the ORC format by itself gives a Spark job no transactions. Avro is row-based and resolves reader and writer schemas, which suits record-at-a-time writes, Kafka payloads via from_avro and to_avro, and landing zones. The usual split: Avro at ingest, Parquet (or a table format over it) for analytics.

Every file has a fixed cost regardless of size: an open, a footer read, object-store requests, and an entry in a listing or a table log. The draft's 'one task per file' is not what Spark 3.5.7 does. File scans pack files into partitions up to maxPartitionBytes, charging openCostInBytes (4 MB by default) per file, so 400 tiny local files became 13 scan partitions. The damage is overhead: in one local run, three filtered counts took 2.89 s over 400 files versus 0.46 s over the same rows in one file. 'No effective pushdown' is also too strong. The plan still showed PushedFilters; per-file cost simply outweighs what skipping saves. Fix it at write time by sizing output partitions to the data, then compact: OPTIMIZE for Delta, or a rewrite into fewer, larger files for plain Parquet.

A plain df.write.parquet(path) or save(path) only writes files. There is no catalog entry, so nothing can be dropped; you delete the directory yourself. df.write.saveAsTable(name) writes under spark.sql.warehouse.dir and registers a managed table: the catalog owns metadata and data, and DROP TABLE removes the table directory. df.write.option("path", p).saveAsTable(name), or CREATE TABLE ... USING parquet LOCATION p, registers an external table: DROP TABLE removes only the metadata, the files stay, and you can register them again. DESCRIBE TABLE EXTENDED shows MANAGED or EXTERNAL in its Type row. The choice is about ownership. Managed fits data whose lifecycle the catalog should control; external fits data shared with other engines or that must survive an accidental drop.

A session built with SparkSession.builder without enableHiveSupport() uses spark.sql.catalogImplementation=in-memory. Table metadata lives only in that application, but the managed table's files stay in the warehouse directory (spark-warehouse by default). The next run starts with an empty catalog, so the table is missing, and a managed table with the same name cannot be created because its default location is not empty. The guide's note that saveAsTable creates a local Derby metastore describes a Hive-enabled session. The pyspark shell tries Hive support by default when Hive classes are available, which is why the same code can behave differently there. Choose the catalog deliberately: Hive support with a stable metastore and warehouse, an external table you register on each run, or path-based reads and writes. Deleting the leftover directory is acceptable only for scratch data.

The table is whatever the transaction log says it is. Each commit is a JSON file in _delta_log named by the next version, listing add and remove actions for immutable Parquet files. Readers load the latest checkpoint plus later commits, so a query sees one complete version and never a half-written one. Writers use optimistic concurrency: read a snapshot, write new files that stay invisible, then create version N+1. If another writer got there first, the loser checks the new commits for conflicts and either retries or fails with an exception such as ConcurrentAppendException. The draft's 'atomically creating the next log file' needs a storage caveat: that atomicity comes from the storage system's mutual exclusion. HDFS and Azure provide it. The Delta 3.2.1 docs say S3 does not, so writes from multiple drivers can lose data unless every writer uses the DynamoDB-backed S3DynamoDBLogStore.

In Delta 3.2.1 a MERGE with matched clauses first inner-joins source and target on the merge condition to find touched files, those with at least one match. It then reads those files again, joins them with the source, and writes updated rows, inserted rows, and every unmodified row from the touched files, before one commit swaps the files. 'Two passes' is right, but the cost driver is rewrite volume: one match in a large file rewrites the whole file. Locally, three matches spread across three 25-row files showed numTargetFilesRemoved 3 and numTargetRowsCopied 72. With deletion vectors enabled, the same MERGE removed 0 files, copied 0 rows, and added 3 deletion vectors. Reduce the cost with a partition predicate in the ON clause, clustering on the merge key so matches land in few files, compaction, and deletion vectors where every reader supports them.

Partitioning splits the table into directories by value, so a filter on that column removes whole partitions from the file list. It fits low-cardinality columns that nearly every query, DELETE, and MERGE filters on, such as a date, and partition predicates also keep concurrent writers apart. On a high-cardinality column it creates masses of tiny files. Z-ordering (OPTIMIZE ZORDER BY) rewrites files so rows with nearby values share files, narrowing each file's min/max range so Delta's file statistics can skip on high-cardinality columns. Locally, a user_id range filter read 1 of 7 Z-ordered files versus 8 of 8 compacted but unordered files. Caveats: statistics cover only the first 32 columns by default, locality weakens with each extra column, and Z-order is not idempotent. A common layout combines both: partition by date, Z-order within partitions.

OPTIMIZE bin-packs small files into larger ones, 1 GiB by default, in a commit that removes the old files and adds the new ones. Readers on older snapshots are unaffected, and plain compaction is idempotent. OPTIMIZE deletes nothing from storage. VACUUM does: it deletes files that no version inside the retention window still needs. In practice, that means files the current version does not reference that became unreferenced more than the threshold ago, 7 days by default. That is more precise than 'unreferenced files past retention': a file dropped an hour ago stays. The cost is history, because time travel to a version whose files are gone fails. Locally, VACUUM RETAIN 0 HOURS was refused by the retention check. With the check disabled, reading version 0 raised SparkFileNotFoundException and RESTORE reported missing files. Log files expire separately, after 30 days by default.

Delta uses optimistic concurrency, so what matters is whether the later commit read or removed files that the earlier commit changed. Per the Delta 3.2.1 docs, blind INSERTs cannot conflict with each other or with compaction. UPDATE, DELETE, and MERGE can conflict with every operation type, and compaction can conflict with those and with other compaction. Locally, three rounds of concurrent INSERTs all committed. An UPDATE racing a DELETE on one unpartitioned table failed with ConcurrentAppendException, because the winner added a rewritten file the loser's whole-table read should have seen. Other failures are ConcurrentDeleteReadException, ConcurrentDeleteDeleteException, MetadataChangedException, and ProtocolChangedException. Partition on the column the jobs filter by and put that predicate explicitly in each condition so their file sets are disjoint. Otherwise, serialize the writers or retry idempotent jobs.

VACUUM removed that version's data files. Time travel needs both the log entries and the data files. VACUUM deletes files that fell out of the current version longer ago than the retention threshold, 7 days by default, while log files expire separately after delta.logRetentionDuration, 30 days by default. Locally, after VACUUM RETAIN 0 HOURS with the safety check disabled, version 0's log was intact. count() returned 10 from the statistics in the log, with a LocalTableScan plan and no Parquet read. collect() on the same version raised SparkFileNotFoundException, and RESTORE refused because not all of version 0's files were available. The retention check exists for this and to protect in-flight writers and long readers. If you need a longer recovery window, raise delta.deletedFileRetentionDuration and delta.logRetentionDuration together.

Writes to Delta are schema-enforced: an append with an extra column fails with a schema-mismatch AnalysisException, and appends evolve only with mergeSchema. MERGE instead follows spark.databricks.delta.schema.autoMerge.enabled, false by default. Locally, with it off, whenMatchedUpdateAll and whenNotMatchedInsertAll succeeded but ignored the new tier column, leaving the target as id and name with no error. An explicit tier = s.tier assignment failed with DELTA_MERGE_UNRESOLVED_EXPRESSION. With autoMerge on, the same MERGE added tier, filled it for updated and inserted rows, and left it null in untouched rows. The trap is the silent drop with star clauses. If a new column matters, enable evolution deliberately for that job or add the column with ALTER TABLE first; do not rely on a session flag set somewhere else.

With delta.enableChangeDataFeed set to true, Delta records row-level changes per commit. A batch or streaming read with readChangeFeed and a startingVersion returns the rows plus _change_type, _commit_version, and _commit_timestamp. The change types are insert, delete, update_preimage, and update_postimage. Locally, updating one row returned both its old and new values in version 1, and a delete returned a delete row in version 2. For UPDATE, DELETE, and MERGE, Delta writes extra change files under _change_data, and two appeared locally. For insert-only commits it can derive changes from the log's add actions instead. The costs are extra writes on DML, and the feed follows the table's retention, so VACUUM deletes change files too. A consumer that falls behind the retention window cannot replay.

The default micro-batch engine turns each trigger into a small batch job: the driver picks an offset range, logs it, plans an incremental Spark SQL query, and commits after the sink finishes. The guide quotes latencies as low as about 100 ms with exactly-once guarantees. Continuous processing, still experimental in Spark 3.5.7, runs long-lived tasks that read and write continuously and checkpoints progress at the trigger interval. The guide quotes about 1 ms, but only at-least-once guarantees. It accepts only map-like operations (projections and filters; no aggregations, current_timestamp, or current_date), the Kafka and rate sources, and the Kafka, memory, and console sinks, and failed tasks are not retried automatically. Locally, a rate-to-memory filter ran continuously, while an aggregation and a file source were rejected with AnalysisException.

Through replay, not one atomic commit. Before a micro-batch runs, the driver writes its offset range to offsets/N in the checkpoint; stateful tasks write a new state version; the sink writes output; only then does the driver write commits/N. After a crash, an offsets entry without a matching commit means batch N is re-run over exactly the same range. That yields exactly-once output only if the source can replay that range (files still present, Kafka offsets still retained) and the sink tolerates the re-run. The file sink is documented as exactly-once, while the Kafka and foreach sinks are at-least-once. In a local crash test, a foreachBatch Parquet append duplicated the replayed batch; a Delta write keyed by txnAppId and txnVersion did not.

At the end of each micro-batch the engine takes the maximum event time seen and subtracts the withWatermark delay; the result is the watermark for the next batch. With several inputs the global watermark is the minimum by default. It follows event time, not the wall clock, so it stalls when data stops. The guarantee is one-directional: rows within the delay are never dropped, and older rows may or may not be. For a windowed aggregation the test is the window's end, not the row's timestamp. Locally, with the watermark at 12:12 and 5-minute windows, a 12:11:30 row was still counted because its window ends at 12:15, while a 12:09 row was dropped and counted in numRowsDroppedByWatermark. Watermarks evict aggregation state only in append or update mode.

Append writes only rows that will never change, update writes rows that changed in this batch, and complete rewrites the whole result table every batch. Which modes are legal depends on the query. Queries without aggregation support append, where update behaves the same, and reject complete. An aggregation without a watermark supports update and complete; append raises an AnalysisException because counts can still change. With an event-time watermark, append works but emits each window once, after the watermark passes its end, so output lags by the delay. Complete mode keeps every aggregate in state forever. Stream-stream joins support append only. Pick append for file-like sinks that cannot rewrite rows, update for upsert-capable sinks, and complete only for small result tables.

metadata holds the query id that survives restarts. offsets/N is the write-ahead record of each batch's source offsets, plus the watermark, batch timestamp, and pinned settings such as the shuffle-partition count and state store provider. commits/N marks batches whose output finished. sources/ holds source-specific logs, such as which files the file source took, and state/ holds versioned state per operator and partition. A restart reuses all of it, so the new plan must match what was stored, and Spark compares state schemas by field count, types, and nullability, not names. Locally, switching count to sum and adding a grouping key both failed with StateSchemaNotCompatible, but swapping one string key for another ran silently and reused the old keys' counts. The guide lists any stateful schema change as not allowed.

With no trigger, the next micro-batch starts as soon as the previous one finishes. processingTime runs batches on a fixed interval; an overrunning batch is followed immediately, and no new data means no batch. availableNow processes everything available at start, possibly across several batches that respect source limits such as maxFilesPerTrigger. It finishes uncommitted batches first, advances the watermark per batch with a final no-data batch, then stops, which suits scheduled incremental jobs. once does the same in a single batch and is deprecated in the 3.5.7 guide in favor of availableNow. continuous selects the experimental low-latency engine with at-least-once guarantees; its interval is a checkpoint interval, not a batch size. Locally, each availableNow run on one checkpoint read only newly added files.

Start with a stream-static left join from the order stream to a batch read of the Delta dimension. It is stateless, and the static side is re-planned in every micro-batch. What that means depends on the static DataFrame. Locally, with another application updating the table between batches, an uncached Delta read returned the new version in the next batch. A cached Delta or Parquet DataFrame kept serving old rows, a plain Parquet directory read kept its original file list, and overwriting that directory made the query fail on a missing file. Either way you get the dimension as of processing time, so a late event joins the current row. For event-time correctness, join an SCD Type 2 table on key and validity range. A watermarked stream-stream join is the alternative when changes arrive as a stream, but it is append-only and can match several versions.

dropDuplicates on the event id alone must remember every id forever, because Spark cannot know when a duplicate stops arriving. Adding the event-time column to the key lets the watermark bound state, but then copies with different timestamps are not duplicates. dropDuplicatesWithinWatermark, new in Spark 3.5.0, requires a watermark and deduplicates on the id alone while copies fall within the delay, then evicts the id. Locally, with a 10-minute delay, id a at 12:00 and again at 12:03 was emitted once; after data at 12:30 moved the watermark to 12:20, two ids were evicted and a copy of a at 12:31 was emitted again. Plain dropDuplicates suppressed that copy, but its state never shrank. Set the delay longer than the largest timestamp gap between duplicates.

foreachBatch is at-least-once by default. The driver writes commits/N only after your function returns, so a crash after your write but before that commit re-runs batch N with the same batchId and, for a deterministic query, the same rows. A plain append writes them twice; locally, a Parquet append ended with order 3 duplicated. Make the write idempotent. For Delta, pass txnAppId (stable for this checkpoint) and txnVersion set to batchId, and Delta skips a transaction it has already committed. Otherwise, MERGE on a business key or overwrite a partition keyed by batch. If the function writes the batch more than once, persist it first: without that, numInputRows for a two-row batch reported 6 because the batch was computed three times.

First check whether state can ever be evicted. In lastProgress, stateOperators reports numRowsTotal, numRowsRemoved, and memoryUsedBytes. A total that only grows means no watermark, a watermark on the wrong column, complete mode, or keys with no time bound. Locally, a plain dropDuplicates never removed a row while the watermarked version removed two. Fix eviction first. The default HDFS-backed provider keeps state in executor JVM memory, so millions of keys mean GC pressure. Setting spark.sql.streaming.stateStore.providerClass to RocksDBStateStoreProvider moves state into native memory and local disk and still checkpoints it; locally its state directory held versioned zip files and SST files, and customMetrics exposed RocksDB counters. Choose the provider before the first run: the guide says it must not change on an existing checkpoint.

Compare several entries of recentProgress, not one. If processedRowsPerSecond stays below inputRowsPerSecond, or durationMs.triggerExecution is longer than the trigger interval, batches are growing. For sources that report it, a latestOffset ahead of endOffset means data is waiting; Kafka reports offsets per topic partition. Then locate the time. durationMs splits into latestOffset and getBatch (source discovery), queryPlanning, walCommit and commitOffsets (the offset and commit logs), and addBatch (running the batch and writing the sink). A large addBatch points at processing or the sink, a growing stateOperators numRowsTotal points at state, and slow commits point at checkpoint storage. Locally the file source reported latestOffset as null, and a continuous query reported no progress at all.

For a classic PySpark batch application, submission and deploy mode determine where the driver runs. Driver-side code creates or reuses a SparkSession and SparkContext, and cluster integration obtains executors that register with the driver. DataFrame operations describe a logical query; Spark SQL analyzes it and, when needed, optimizes and plans physical execution. An action requests results and can trigger multiple jobs. The DAG scheduler separates execution dependencies at shuffle boundaries and creates tasks for runnable stage partitions. The task scheduler places task attempts on available executor resources through its backend. Executors run operators, may partly aggregate, and exchange data when the plan requires it. AQE can adjust downstream execution using runtime statistics. Status and metrics return to the driver, recoverable failures can be retried, and the output is returned to the driver or written to storage. The application can then run more work or shut down.

Not necessarily. A user-level SQL action is not a promise of one scheduler job. Broadcast preparation, adaptive execution, or incremental result retrieval can lead to extra work. Start with the SQL execution and its linked jobs, then read the executed plan.

In a CPU-only model with all 12 executor cores available, each four-core executor offers floor(4 / 2) = 2 task slots, so six task attempts can run concurrently. Ready tasks, other resource requirements, and competing work can reduce that number.

No. A shuffle exchange redistributes data for a required partitioning. A BroadcastExchange prepares and distributes a broadcast relation. Read the operator type and the surrounding join or aggregation, rather than counting the word Exchange.

Where the driver JVM runs. In client mode the driver lives on the machine that submitted the application — handy for interactive work, and fatal to the job if that machine goes away. In cluster mode the cluster manager launches the driver inside the cluster, so the submission client can exit and the application keeps running. The driver's location also decides where driver-side memory pressure and logs land.

The filter is local to each input partition, but the final count is global. Under this four-partition baseline I expect four tasks to filter and partially count, then a single downstream task to combine those counts. The answer is nine. I would verify the count query, rather than only explaining the filtered DataFrame.

The aggregation needs compatible distribution by bucket, so I expect partial sums, a shuffle, and final sums. With AQE disabled and four shuffle partitions, I predict four downstream writer tasks even though there are only three keys. The Parquet call triggers the work, and tasks write to storage while the driver coordinates it. File count needs separate inspection.

Broadcasting the tiny lookup lets the fact-side tasks perform this join without reshuffling orders for the join. The subsequent region aggregation still requires a shuffle. I would distinguish the broadcast exchange from the hash exchange in the plan, check the associated jobs, and validate that the lookup keys do not multiply rows.

Before count(), the reader already starts work. With header and inferSchema, Spark 3.5.7 runs a one-task job for the first line and a second job that scans all four input partitions to infer types. count() is a global aggregate: four scan tasks produce partial counts, a single-partition shuffle moves them, and one task adds them. With AQE off that is one job with two stages. With AQE on, the shuffle stage runs as its own job and the result job shows it as skipped. The Parquet write has no shuffle, so it is one stage of four tasks. On this input that was four jobs and 14 tasks with AQE off, five jobs with AQE on. Supplying a schema removes both read jobs.

Reading Parquet without mergeSchema ran one small footer job. The filtered show() has no shuffle. show() asks for 21 rows, Spark scans one partition first, and that partition had enough, so one job with one task. The SQL aggregation adds a hash shuffle on user_id. With AQE off there are 200 shuffle partitions but only 40 users, so show() scanned them in rounds of 1, 4, 20, 75, and 8 partitions. That was five jobs, and the last four each listed the four-task map stage as skipped. With AQE on, the map stage ran as its own job, AQE coalesced the tiny output to one partition, and one more job finished it. The number of show() jobs depends on where rows land, so I would measure it.

There is one shuffle, not two. repartition(10, customer_id) hash-partitions by the grouping key, which already satisfies the aggregate’s requirement, so the plan has one Exchange and both aggregates run after it. The read has header but no inferSchema, so it is a single one-task job. The shuffle has four map tasks, one per file, and ten reduce partitions. AQE does not coalesce a repartition with an explicit number. show() needs 21 rows at about five per partition, so it scanned 1, then 4, then 2 partitions. With AQE off that is three jobs and 11 tasks. With AQE on, the map stage ran as its own job, so four jobs with the same tasks.

Two reads come first: each Parquet table ran a one-task footer job. customers is tiny, so the default plan broadcasts it. The broadcast collection is its own one-task job, and show() then probes a single orders partition, so there is no shuffle. If I disable broadcast, the plan becomes a sort-merge join with two hash shuffles. With AQE off, one job has three stages: four map tasks for orders, one for customers, and one reduce task, because the first reduce partition already had 21 joined rows. With AQE on, each shuffle stage runs as its own job and the result job lists both as skipped. Same tasks, three jobs. Broadcasting avoids the shuffles, but it still adds a job.

cache() does not start anything by itself, so the only job before the first count is the Parquet footer read. With AQE off, the first count is one job with two stages. Four tasks scan the files, fill the cache, and compute partial counts, and one task adds them. The filtered count is again four plus one tasks, but its plan reads InMemoryTableScan and its input metrics are cached blocks, not Parquet bytes. With AQE on, Spark 3.5.7 builds the cache as its own query stage first, so the first count ran three jobs: build the cache, the shuffle map stage, then the result. The filtered count ran two. Caching saves rereading the files, not tasks.

One action, collect, so one job. groupByKey adds a shuffle, so there are two stages, and mapValues runs inside the second one. The first stage has three tasks, one per parallelize partition. In PySpark, groupByKey without a partition count uses spark.default.parallelism only if that key is set; otherwise it keeps the parent’s three partitions, so the reduce stage has three tasks. That makes six tasks. AQE does not apply to RDDs, and I measured the same counts with it on and off. The result groups a to [1, 3], b to [2, 4], and c to [5]. Key order depends on hashing, so I would not promise it.

map and filter are narrow and run together in the same task, and collect is the only action, so this is one job with one stage. The stage runs one task per partition, two tasks. The first partition keeps nothing after the filter, but its task still runs. The result is 12, 14, 16, 18, and 20. AQE does not change RDD jobs, and I measured the same counts either way.

The JSON read infers a schema, which is a four-task job before any action. The aggregation adds a hash shuffle on user_id. With AQE off there are 200 shuffle partitions but only 40 users, so show() scanned in rounds of 1, 4, 20, 75, and 8 partitions: five jobs, and the later ones list the map stage as skipped. The write is a separate action with its own plan. Without a cache it rescans the JSON and reruns the shuffle, then runs 200 write tasks and writes 38 files, since only 38 partitions hold data. With AQE on, each action runs a map-stage job and a result job on one coalesced partition. Nothing is reused between show and write.

Rank rows inside each key and keep the first: row_number over a window partitioned by the key and ordered by the timestamp descending, then keep row number 1. Add a unique tie-breaker such as an ingest ID to the ordering and decide where null timestamps sort, or the survivor of a tie is arbitrary. The window costs one hash shuffle on the key plus a sort within partitions; for a row_number = 1 filter, Spark 3.5.7 inserts a WindowGroupLimit before the shuffle so tasks forward only local winners. An aggregation alternative is max over a struct of (timestamp, tie-breaker, payload), or max_by with a struct as the ordering value. Do not rely on orderBy followed by dropDuplicates: it becomes a first() aggregate after a shuffle, and the sort does not decide which row wins. The full solution is in the coding-round lesson.

Order each user's events and compare neighbors. Over a window partitioned by user and ordered by timestamp, plus an event ID for equal timestamps, lag gives the previous timestamp; flag a new session when there is no previous event or the gap exceeds 30 minutes; a running sum of the flag over a rows frame from unbounded preceding to the current row numbers the sessions. Agree the boundary first: is a gap of exactly 30 minutes the same session? Both window functions share one specification, so the plan is a single shuffle on the user key and one sort; a very active user still lands in one task. For per-session aggregates, session_window (3.2+) ends each session at the last event plus the gap and also works in Structured Streaming, where lag over a window does not. The full solution is in the coding-round lesson.

Define it first: for each date, the number of distinct users with any activity on that date or the six days before it. Spark 3.5.7 rejects a distinct count over a window frame with DISTINCT_WINDOW_FUNCTION_UNSUPPORTED, and summing daily active counts is wrong twice: a rows frame covers seven rows, not seven calendar days, when dates have gaps, and even a range frame on a day number double-counts users active on several days. A scalable approach deduplicates user-days, explodes each into the seven as-of dates it counts toward, groups by as-of date with a distinct count, and left-joins a date spine so quiet days appear as zero. That is a sevenfold row expansion and two aggregation shuffles, with no global sort. A collect_set over a rangeBetween frame is also correct, but an unpartitioned window moves every row into one partition. The full solution is in the coding-round lesson.

A changed key needs an update and an insert in one MERGE, but a source row either matches or it does not. Stage the source as a union: every batch row keyed by its business key, plus an extra copy of each changed row with a NULL merge key, found by joining the batch to current dimension rows and comparing attributes null-safely. Merge on key equals merge key and is_current: matched rows whose attributes differ are closed with valid_to and is_current false; unmatched rows, meaning new keys and the NULL-keyed copies, are inserted as current. Unchanged rows match but fail the update condition, so replaying the same batch is a no-op. Deduplicate the batch to one row per key first, or Delta raises a multiple-source-rows error. Cost: the staging join is a shuffle or broadcast, and MERGE searches the whole target by default unless the match condition carries partition predicates. The full solution is in the coding-round lesson.

orderBy followed by limit returns the top N of the entire DataFrame, not of each group: one group can take every slot and the others vanish. Its cost is not the issue, because Spark plans it as TakeOrderedAndProject, a top-N selection rather than a full global sort. Per group, rank inside a window partitioned by the group and ordered by the metric, then filter. The tie rule picks the function: row_number returns exactly N rows but needs a deterministic tie-breaker; rank keeps ties and skips numbers (1, 1, 3); dense_rank keeps ties without gaps, so it can return more than N rows. The cost is one hash shuffle on the group key and a sort within partitions; in Spark 3.5.7 a filter such as rn <= N lets the optimizer insert WindowGroupLimit before and after the shuffle, trimming each group early. The full solution is in the coding-round lesson.

Read with an explicit DDL schema so Spark skips the inference pass and a missing field becomes null instead of changing the schema. Select nested struct fields with dotted paths such as customer.address.city, which are null when a parent struct is null or absent, or expand one level with customer.*. For the array, explode emits one row per element but drops parents whose array is null or empty; explode_outer, posexplode_outer, and inline_outer keep them with null columns. inline turns an array of structs directly into columns, and posexplode keeps each element's position. Generators are narrow, so there is no shuffle, but they multiply rows, and exploding two independent arrays one after the other produces their cross product; explode one array per output, or zip related arrays first. The full solution is in the coding-round lesson.

I would split capture from apply. Reading MongoDB's change stream is a connector's job outside Spark; it should land each event with the document key, the operation, and a source-ordered position in Kafka or files. 500 GB/day averages under 6 MB per second, so micro-batches are comfortable, sized for peak rather than average. Bronze appends raw events unchanged and is retained long enough to replay. Silver applies each micro-batch in foreachBatch: keep the latest event per key by that position with row_number, then a Delta MERGE whose update fires only when the incoming position is higher, so late or replayed events cannot regress state. Deletes become tombstone rows rather than hard deletes, so a late update cannot resurrect a key; a current view filters them. Documents drift, so Bronze keeps the raw payload, Silver adds fields deliberately with an alert, and type conflicts go to quarantine. Replay reruns apply from Bronze and converges.

No code change does not mean nothing changed, so I compare a good run with the slow one instead of tuning blind. First, input: did bytes, rows, or file counts grow, or did an upstream start writing thousands of small files? Second, in the Spark UI, find the stage whose duration grew most and compare its max task time and shuffle read with the median: one huge task points to a new hot key, uniformly slower tasks point to volume. Third, diff the SQL plans: a join that used to broadcast may now be sort-merge because a side grew past the threshold or statistics changed, and AQE may have chosen differently at runtime. Fourth, the environment: fewer or smaller executors, a changed Spark or library version, or a config drift. Fifth, spill and GC time, plus failed tasks and retried stages from lost executors or fetch failures. Then change only what the evidence points to and rerun.

I would first pin down the requirement: source event to queryable row, at what percentile, and what a miss costs. Five minutes is well within Structured Streaming's micro-batch engine, which the Spark docs describe as reaching latencies as low as about 100 milliseconds, so experimental continuous processing, which is also at-least-once, is unnecessary. My default is a micro-batch query with a processing-time trigger around one minute, a checkpoint, and an idempotent Delta sink, which leaves headroom for a slow batch. A cron-scheduled batch every few minutes also works, but you then own incremental bookkeeping that the checkpoint provides, and startup time eats the budget. If always-on compute is too expensive, a scheduled Trigger.AvailableNow run reuses the same streaming code and checkpoint in batch-style runs, provided interval plus runtime stays under five minutes. Either way I would alert on end-to-end lag, not on job success.

I separate row-level failures from batch-level ones. Row rules such as a null key or a negative amount split the batch: valid rows publish, failing rows go to a quarantine table with the rule name and batch ID, and the quarantine rate is a metric with an alert threshold. True invariants go into the table as NOT NULL and CHECK constraints; in Delta one violating row fails the entire write, which is loud but blocking, so I use them sparingly. Batch-level checks — row count against a trailing baseline, duplicate keys, freshness — use write-audit-publish: write to staging, run the audits, then publish in one commit. A failed audit pages someone while consumers keep reading the last good version instead of half-checked data. Finally, I alert on silent success: a job that writes zero rows, or far fewer than usual, should page as loudly as a crash.

I would start by separating what Spark enforces from what the platform enforces. Spark 3.5.7 provides authentication, encryption, and Web UI view and modify ACLs, none enabled by default, but no table-, row-, or column-level permissions on data. Those come from a catalog or governance layer such as Unity Catalog, Apache Ranger, or AWS Lake Formation, each with its own vendor- or project-specific model. The design: tenants as groups rather than users; grants at schema or table level by default; row filters or per-tenant views where tenants share a table; column masks for sensitive fields; and grants managed as reviewed code. Then close the bypass: a job that reads the object store directly is not stopped by a SQL grant, so storage credentials must be scoped to match. Finally, audit access logs and test that one tenant's queries return nothing belonging to another.

Make the partition, not the run, the unit of truth: a backfill replaces a partition's contents instead of appending to it. For each date, recompute from the immutable source and overwrite just that date. On Delta, replaceWhere on the date predicate is atomic and fails if any row falls outside it. On plain Parquet or other file-source tables, set partitionOverwriteMode to dynamic, because the default static mode on an unscoped overwrite deletes partitions you did not touch. Run dates in bounded chunks so a failure retries one chunk, and keep the live job from writing the same partitions meanwhile. Two traps: an append guarded by a date-derived txnVersion silently skips older dates, and dynamic overwrite replaces a whole partition if one stray row lands there. Recompute downstream aggregates for the same range, and reconcile per-date counts before and after.

Contain, scope, repair, prove, prevent. Contain: fix or pause the writer and stop downstream jobs from propagating more. Scope: use table history to find the first bad version and which partitions and downstream tables consumed it. Repair Bronze: if nothing good was written since, RESTORE to the last good version, which writes a new, auditable commit; if good data is interleaved, rewrite only the affected days from the raw landing copy with replaceWhere. RESTORE works only while old data files survive VACUUM, seven days by default, so check retention before promising it; after VACUUM, replay from immutable raw input is the only path, and if the source sent bad data, the source must resend. Then recompute Silver and Gold for the affected partitions idempotently, reconcile counts against the source, and add the check that would have caught it on day one.

Fill this in with a real incident of your own; this card gives only the structure an interviewer probes. Situation: the job, its schedule, input size, and cluster size. Task: the SLA miss as a number, such as stage runtime against target. Action: evidence first. From the Spark UI, the slow stage's max task duration and shuffle read against the median, then the hot key and its share of rows, and whether AQE skew-join handling applied; in Spark 3.5.7 it splits skewed partitions of sort-merge and shuffled hash joins and does not fix a skewed aggregation. Then the fix you chose, such as salting, isolating hot keys into a broadcast join, pre-aggregating, or filtering null keys, and what you rejected. Result: before and after stage time, total runtime, and cost, measured the same way on comparable input. Expect a follow-up on why adding executors would not have helped.

Use a real incident of your own; this card gives the shape. Situation: the pipeline, how often it runs, data volume per run, and cluster. Task: the baseline in one unit you can defend, such as runtime per run, node-hours, monthly cost, or bytes scanned, and why it mattered. Action: where the time or money went before you changed anything, from the Spark UI and SQL plan: the dominant stage, scans reading far more than the query needed, shuffle volume, spill, idle executors, or small files. Then the one or two changes with the largest effect, such as pruning partitions, fixing file layout, removing an unneeded shuffle or cache, broadcasting a small side, or right-sizing the cluster, and the trade-off each carried. Result: before and after in the same unit on comparable input, the percentage, and whether it held over later runs. Interviewers probe attribution, so know how you proved the change caused the saving.

Fill this in with an incident you actually handled; this card gives the structure. Situation: the table, its consumers, and what they used it for. Task: the discrepancy as numbers — rows or amounts off, by what percentage, over how many days — and how it surfaced: a reconciliation, a consumer, or an alert. Action: the root cause in one sentence and how you proved it, for example a non-idempotent append that double-counted on retry, updates applied in arrival order instead of source order, a join that fanned out on duplicate keys, or a timezone boundary. Then the repair of history, such as a restore, a replaceWhere backfill, or a replay from raw, and how you told consumers. Result: counts reconciled after the fix and the guard you added, such as a uniqueness check, a sequence-guarded merge, or a row-count audit. Expect probing on blast radius and on why the pipeline still reported success.