Apache Spark Theoretical

The reasoning

Lazy evaluation means a transformation such as `filter`, `select` or `join` returns a new DataFrame that describes work, not data. Nothing is read and nothing is computed. Spark adds the step to a logical plan and waits.

An action such as `count()`, `collect()`, `show()` or `write` forces the plan to run. At that moment Catalyst sees the whole chain. It can push a filter below a join, drop columns nobody reads, and fold constants, because it knows the end before it starts.

For a pipeline this changes two habits. First, you can build a plan in small named steps without paying for each one. Second, every action re-runs the plan from the start unless you cache, so two actions on the same DataFrame read the source twice.

The cost is that errors surface late. A bad column name fails at the action, not at the line that named it. `explain()` shows the plan without running it, which is the cheap way to check.

The reasoning

A transformation is a description. `df.filter(col('amount') > 100)` returns a DataFrame whose plan ends in a Filter node. You can chain hundreds of transformations and no task runs.

An action is a request for output. `count()` returns a number to the driver, `collect()` returns all rows to the driver, `show()` prints a few, and `write` puts files on storage. Each action submits a job, and Spark divides that job into stages and tasks.

A few calls look like actions but are not. `cache()` and `persist()` only mark the DataFrame. `explain()` runs the optimizer but no tasks. `createOrReplaceTempView` registers a name. The rule is simple: if nothing leaves the cluster and no file is written, it is not an action.

Transformations split into narrow and wide. Narrow ones such as `filter` keep each input partition on its own. Wide ones such as `groupBy` and most joins need a shuffle, and a shuffle is where a stage ends.

The reasoning

Inside a job, a DataFrame is split into partitions. Each partition is processed by exactly one task, so a stage with 200 partitions runs 200 tasks. More partitions means more parallel work, up to the number of cores you have.

On disk, a partitioned table is a folder tree. `partitionBy('dt')` on write creates one folder per date. A query with `WHERE dt = '2026-11-01'` reads only that folder. This is partition pruning, and it has nothing to do with task count.

The two meet at read time. Spark turns the files it must read into Spark partitions, using file sizes and `spark.sql.files.maxPartitionBytes`. A table with 10 date folders might become 300 Spark partitions, or 10, depending on file sizes.

Keep the vocabulary apart in an answer. Say "shuffle partitions" or "task partitions" for the in-memory kind, and "table partitions" or "Hive-style partitions" for folders.

The reasoning

For file sources such as Parquet, ORC, JSON and CSV, Spark lists the files and their sizes. It then packs them into partitions. No partition is bigger than `spark.sql.files.maxPartitionBytes`, which the configuration page documents as 134217728 bytes, 128 MB. A 1 GB splittable file becomes about 8 partitions.

Small files are bundled. Each file is counted as its size plus `spark.sql.files.openCostInBytes`, documented as 4194304 bytes, 4 MB. The page says to over-estimate this cost on purpose, so that many small files are grouped into fewer tasks rather than one task each.

So a folder of 10,000 files of 100 KB each becomes a few hundred partitions, not 10,000. And a single 5 GB file becomes about 40 partitions, as long as the format can be split. Gzip-compressed text cannot be split, so one file is one partition however large it is.

The maximum degree of parallelism is also bounded by the number of cores, through `spark.default.parallelism` and the cluster size. After the first shuffle, the count is set by `spark.sql.shuffle.partitions` instead.

The reasoning

Every shuffle writes its output into this many partitions. The SQL performance tuning page documents the default as 200. A 5 TB join with 200 partitions gives 25 GB tasks that spill or fail. A 50 MB aggregation with 200 partitions gives 200 tiny tasks that spend their time starting up.

The sizing rule is to divide the expected shuffle bytes by a target partition size. With 1 TB after the shuffle and a 200 MB target, that is about 5,000 partitions. Round to a multiple of your core count so the last wave of tasks is not half empty.

Since Spark 3.2, adaptive query execution is on by default, and `spark.sql.adaptive.coalescePartitions.enabled` merges small shuffle partitions at runtime towards `spark.sql.adaptive.advisoryPartitionSizeInBytes`, documented as 64 MB. So the practical approach is to set the number high enough for the biggest shuffle and let AQE shrink the small ones.

This setting applies to DataFrame and SQL shuffles only. RDD operations such as `reduceByKey` use `spark.default.parallelism` instead, which is a frequent source of confusion.

The reasoning

A broadcast hash join sends a copy of the small side to every executor and builds a hash table there. No shuffle happens on the large side. Spark picks it when the small side is estimated below `spark.sql.autoBroadcastJoinThreshold`, which the tuning page documents as 10485760 bytes, 10 MB. Setting it to -1 disables it.

A sort-merge join shuffles both sides by the join key, sorts each partition, and walks them together. It handles any size on either side at the cost of a full shuffle and a sort. It is the default for an equi-join that cannot broadcast.

A shuffle hash join also shuffles both sides but builds a hash table per partition instead of sorting. Spark prefers sort-merge unless `spark.sql.join.preferSortMergeJoin` is false or a hint asks for it. The nested loop joins are the last resort for joins with no equality condition.

You can steer the choice with hints: `BROADCAST`, `MERGE`, `SHUFFLE_HASH` and `SHUFFLE_REPLICATE_NL`. When the two sides carry different hints, the docs rank them: broadcast beats merge, which beats shuffle hash. A hint is not a guarantee, because a strategy may not support the join type.

The reasoning

After `groupBy` or a join, rows with the same key land in the same partition. If `country = 'US'` is 60 percent of the data, one partition holds 60 percent of the shuffle. That one task does most of the work while the other tasks sit finished. The stage takes as long as its slowest task.

To see it, open the stage in the Spark UI and sort the task table by duration or by shuffle read size. Skew looks like 199 tasks at 2 seconds and 1 task at 25 minutes, with shuffle read of 8 GB against 200 MB. The summary metrics at the top show the same thing as a max far above the 75th percentile.

There are four fixes, in the order to try them. Broadcast the small side so no shuffle is needed. Let AQE split the skewed partition, which is on by default for sort-merge joins. Salt the key: add a random suffix on the big side and explode the small side to match. Or handle the hot key in a separate job.

Skew is a property of the data, not the code, so it comes back when the data changes. A job that was fine for a year can start failing the month one customer grows.

The reasoning

Static planning uses estimates, and estimates are often wrong. AQE waits for a shuffle stage to finish, reads the actual output sizes, and re-optimizes the rest of the plan. `spark.sql.adaptive.enabled` is documented as true by default.

First, coalescing. When `spark.sql.adaptive.coalescePartitions.enabled` is true, Spark merges adjacent small shuffle partitions until each is close to `spark.sql.adaptive.advisoryPartitionSizeInBytes`, documented as 64 MB. That is why 200 tiny tasks become a handful.

Second, skew. With `spark.sql.adaptive.skewJoin.enabled`, a partition is treated as skewed when it is both larger than `skewedPartitionFactor` times the median, default 5, and larger than `skewedPartitionThresholdInBytes`, default 256 MB. Spark splits it into pieces and joins each piece separately.

Third, join conversion. If one side of a sort-merge join turns out to be below the broadcast threshold after its shuffle, AQE switches to a broadcast hash join. The local shuffle reader then avoids a second network pass. In `explain()`, an adaptive plan shows `AdaptiveSparkPlan` with `isFinalPlan=false` until the query has run.

The reasoning

Both are RDD operations on pairs. `groupByKey` returns `(key, Iterable[values])` and has to move every value to the partition that owns its key. For a key with 100 million values, one task receives all of them, and the whole list is held in memory at once.

`reduceByKey(func)` runs `func` locally first, like a map-side combine. If a partition holds 1 million rows for `US`, it reduces them to one partial sum before anything moves. The shuffle carries one row per key per partition, and the receiving task adds a few partials.

The practical rule is that `groupByKey` is only right when you genuinely need the full list of values per key. For any aggregation that can be expressed as a merge, use `reduceByKey` or `aggregateByKey`. On DataFrames, `groupBy().agg()` already does the partial aggregation for you.

This is also why a DataFrame `groupBy` is usually safe where an RDD `groupByKey` runs out of memory: Catalyst inserts a partial aggregation before the exchange.

The reasoning

Take `spark.executor.memory` as the heap. The tuning guide reserves 300 MiB, then gives `spark.memory.fraction` of the rest, default 0.6, to the unified region M. The other 40 percent is user memory for your own objects, Spark's internal metadata, and a safety margin against large records.

Inside M, execution memory holds shuffle buffers, join hash tables and aggregation maps. Storage memory holds cached and broadcast blocks. The two share M. Execution can evict cached blocks, but only down to R, which is `spark.memory.storageFraction` of M, default 0.5. Storage can never evict execution.

Outside the heap, `spark.executor.memoryOverhead` is extra memory for the JVM itself, native buffers, and Python workers. The configuration page documents it as executor memory times `spark.executor.memoryOverheadFactor`, default 0.10, with a minimum. On Kubernetes with non-JVM jobs the factor defaults to 0.40, which is why PySpark pods need more.

Worked numbers for a 20 GB executor: about 19.7 GB after the reserve, and 11.8 GB in M. Of that, 5.9 GB is protected for storage. About 7.9 GB is user memory. The container asks for 22 GB because of the 10 percent overhead.

The reasoning

Both calls mark a DataFrame to be kept after the first action computes it. `persist(level)` lets you choose how. The RDD guide lists `MEMORY_ONLY`, `MEMORY_AND_DISK`, `MEMORY_ONLY_SER`, `MEMORY_AND_DISK_SER`, `DISK_ONLY`, the `_2` variants that replicate each partition on two nodes, and the experimental `OFF_HEAP`.

`MEMORY_ONLY` stores deserialized objects and recomputes any partition that does not fit, every time it is needed. `MEMORY_AND_DISK` spills the partitions that do not fit to local disk instead. The serialized levels store one byte array per partition, which is smaller but costs CPU to read. In Python, the guide notes, objects are always pickled, so the serialized flag makes no difference.

Cache only when the same DataFrame feeds two or more actions and the recompute is expensive. A single-use DataFrame gains nothing and pays for the write into memory. A DataFrame that does not fit evicts other blocks and still recomputes, so the job gets slower with the cache than without it.

Release with `unpersist()` when the stage that needed it is over. The Storage tab in the Spark UI shows what is cached, what fraction is in memory, and what spilled to disk.

The reasoning

Spark serializes objects when they cross the network in a shuffle, when they are cached in a serialized storage level, and when a closure is sent to executors. The default `spark.serializer` is `org.apache.spark.serializer.JavaSerializer`, which is flexible and slow.

Setting `spark.serializer` to `org.apache.spark.serializer.KryoSerializer` switches to Kryo. The tuning guide says to register your classes with `registerKryoClasses`, because an unregistered class is written with its full class name on every object, which is wasteful. Large objects may need a bigger `spark.kryoserializer.buffer`.

DataFrames mostly bypass this. Tungsten stores rows in its own binary `UnsafeRow` format, and a DataFrame shuffle moves those bytes, not Java objects. So Kryo helps RDD jobs, serialized caches and custom objects inside UDFs, and does little for a plain SQL pipeline.

Kryo is already used internally for shuffling RDDs of simple types since Spark 2.0. The switch is for your own classes and for RDD-heavy code.

The reasoning

`java.lang.OutOfMemoryError: Java heap space` on an executor means a task needed more than execution memory could give, and could not spill. A join hash table for one skewed key does that. So does an `explode` that turns one row into millions. Fix the partition size or the skew, not the cluster.

`GC overhead limit exceeded` is the JVM spending almost all its time collecting garbage with little memory freed. It is the same shortage one step earlier. Fewer cores per executor, less caching, or more memory per task all help.

The driver runs out when results come home: `collect()` on a large DataFrame, `toPandas()` without Arrow limits, or a broadcast that is collected on the driver first. Fix the code; raising `spark.driver.memory` only moves the wall.

`Container killed by YARN for exceeding memory limits`, or a Kubernetes OOMKilled with exit code 137, is different. It means the whole process exceeded the container: heap, off-heap and Python workers together. Raise `spark.executor.memoryOverhead`, or on Kubernetes the overhead factor, which defaults to 0.40 for non-JVM jobs for exactly this reason.

The reasoning

The submitting-applications page defines `--deploy-mode` as whether the driver runs on the worker nodes, cluster, or locally as an external client, client. The documented default is client.

In client mode the driver's input and output are attached to the console, which the docs call especially suitable for the REPL. The job dies if the submitting machine disconnects, and `collect()` lands on that machine's memory.

In cluster mode the cluster manager starts the driver on a worker. The docs recommend it when the application is submitted from a machine far from the workers, to keep driver-to-executor latency low. The job survives the submitter closing its laptop, and the driver's memory comes from the cluster.

Two notes from the docs. Standalone mode does not support cluster mode for Python applications. And in cluster mode the application file must sit at a path every node can read, such as `s3://` or `hdfs://`, not on the submitter's disk.

The reasoning

The basic form is `spark-submit --class <main> --master <url> --deploy-mode <mode> --conf key=value <application-jar or .py> [args]`. The docs say the application file must be visible to every node in cluster mode, and that application arguments go after it.

`--master` takes `local[N]`, `yarn`, `k8s://https://host:port` or `spark://host:7077`. `--conf` sets any property; the docs say to pass several as separate arguments and to quote values with spaces. Properties set here override `spark-defaults.conf`, and properties set in code on `SparkConf` override both.

For Python, pass a `.py` file in place of the jar and use `--py-files` to add `.py`, `.zip` or `.egg` files to the search path. `--jars` takes a comma-separated list that is added to driver and executor classpaths; the docs note that directory expansion does not work. `--packages` takes Maven coordinates and resolves transitive dependencies, with `--repositories` for extra repositories.

Resource flags matter most in practice: `--executor-memory`, `--executor-cores`, `--num-executors` on YARN, and `--driver-memory`. Those are the four numbers an interviewer will ask you to size for a given cluster.

The reasoning

The scheduler retries a task that throws or whose executor dies. The configuration page documents `spark.task.maxFailures` as 4, the number of continuous failures of any particular task before giving up on the job. A task that fails four times for the same reason, such as a bad record in a UDF, takes the whole job down.

If the executor was lost, its finished shuffle output is lost too, so the stages that produced it may be recomputed. That is why one lost executor late in a job can cost more than the task it was running.

Speculation is different. `spark.speculation` is false by default. When it is on, Spark checks every `spark.speculation.interval`, 100 ms, for slow tasks. A task counts as slow when it has run `spark.speculation.multiplier`, 3, times longer than the median. The check starts once `spark.speculation.quantile`, 0.9, of the stage's tasks have finished. It launches a duplicate and keeps the first to finish.

Speculation helps when one node is slow, for example a bad disk or a noisy neighbour. It does not help with skew, because the duplicate has the same huge partition. And it doubles the work of the slow tasks, which costs money on a cloud bill.

The reasoning

PERMISSIVE is the default. When a row cannot be parsed, Spark keeps it and sets the malformed fields to null. If your schema has a string column with the corrupt-record name, the original text is written there. The docs say that without that column the row is still kept but the raw string is not available.

DROPMALFORMED removes the row. It is convenient and dangerous, because nothing tells you how many rows disappeared. The docs note this mode is unsupported in the CSV built-in functions such as `from_csv`.

FAILFAST throws an exception at the first corrupt record. It is the right choice when a bad row means the file is wrong, for example a vendor feed that must be rejected whole.

The production pattern is PERMISSIVE with a `_corrupt_record` column, then a split. Rows where it is null go on. Rows where it is not null go to a quarantine table, with the file name and a count. The column name defaults to `spark.sql.columnNameOfCorruptRecord`, and `inferSchema` is documented as false by default and costs an extra pass over the data.

The reasoning

The docs are blunt about the modes: they do not use locking and are not atomic, and an overwrite deletes the data before it writes the new data. So a failed overwrite can leave an empty table, which is why serious pipelines write to a staging location or use a table format.

`append` is right for immutable events, and wrong for reruns because the second run adds the rows again. `overwrite` is right for recomputing a known slice. It is wrong when the DataFrame holds only part of the table, because the default static mode first deletes every partition that matches the write's partition spec.

`spark.sql.sources.partitionOverwriteMode` is documented with default `STATIC`. In `DYNAMIC` mode Spark does not delete partitions ahead of time and only overwrites the partitions that receive data at runtime. That is the setting that makes `overwrite` with `partitionBy('dt')` safe for rewriting one day without touching the others.

The cloud integration page adds a limit. Dynamic partition overwrite needs a file system with an efficient rename, and the S3A committers do not have one. On S3 the job fails with `PathOutputCommitter does not support dynamicPartitionOverwrite`. Table formats such as Delta and Iceberg replace this mechanism with `replaceWhere` or `MERGE`.

The reasoning

`saveAsTable` registers a table in the metastore, which the docs describe as materializing the DataFrame and creating a pointer to it in the Hive metastore. Without a `path` option, the data goes to a default path under `spark.sql.warehouse.dir`. That is a managed table: when it is dropped, the default path is removed too.

With `df.write.option('path', '/some/path').saveAsTable('t')`, the table is external. The docs say that when the table is dropped the custom path is not removed and the data is still there. The same is true for `CREATE TABLE ... LOCATION` in SQL.

Two consequences follow. For external tables created from existing files, partition information is not gathered by default, and the docs say to run `MSCK REPAIR TABLE` to sync partitions into the metastore. And a plain `save('/path')` registers nothing, so the table exists only as files until someone creates a table over them.

Managed tables are convenient for scratch work. For a lake that other engines read, external tables or a table format with its own catalog are the normal choice, because the data outlives any one engine's table definition.

The reasoning

Without Arrow, moving a DataFrame to pandas pickles every row through the Python worker. With Arrow, the JVM writes record batches in Arrow's columnar layout and Python reads them in place. The PySpark Arrow guide says the optimization applies to `toPandas()` and to `createDataFrame()` from a pandas DataFrame, and that results are identical with or without it.

Batches are bounded by `spark.sql.execution.arrow.maxRecordsPerBatch`, documented as 10,000 rows, to keep JVM memory in check. The guide warns that this limit is not applied to grouped-map operations, where a whole group is loaded at once.

Pandas UDFs are built on the same path. A Series-to-Series UDF takes one or more `pandas.Series` and returns one of the same length; Spark splits the column into batches, calls the function per batch, and concatenates. The iterator variants let you set up state once per partition. `mapInPandas` maps an iterator of pandas DataFrames and may return a different number of rows.

Fallback is controlled by `spark.sql.execution.arrow.pyspark.fallback.enabled`: if Arrow conversion fails before computation, for example on an unsupported type, Spark falls back to the non-Arrow path rather than failing. `toPandas()` is still a `collect()`; Arrow makes it faster, not safe for a 100 GB DataFrame.

The reasoning

`partitionBy('dt')` on write creates `dt=2026-11-01/` style directories. The load-save docs note that it creates a directory structure and therefore has limited applicability to columns with high cardinality. A million `user_id` folders is a listing problem, not a layout.

`bucketBy(42, 'user_id')` distributes rows into 42 files by hash of the key. The docs say bucketing can be used when the number of unique values is unbounded, and that bucketing and sorting apply only to persistent tables, meaning `saveAsTable`, not `save`.

Two tables bucketed on the same key with the same bucket count can be joined without a shuffle, because matching keys are already in matching buckets. That is the one case where a join is narrow. The price is that every writer must bucket the same way, and the bucket count is fixed at creation.

They combine: partition by date, bucket by customer inside each date. Table formats handle this differently; Iceberg has a bucket transform in its partition spec, and Delta prefers clustering over Hive-style buckets.

The reasoning

By default Spark reads the schema from a sample of files, which is fast but assumes the files agree. When a column was added last week, old files lack it and the sampled schema may miss it or include it depending on which files were sampled.

`spark.read.option('mergeSchema', 'true').parquet(path)` tells Spark to open every file's footer and merge the schemas. Added columns appear with nulls in old files. Incompatible type changes, such as an int becoming a string, still fail.

The cost is proportional to the file count, because each footer is a separate read, and on object storage each read is an API call. On a table with 100,000 small files that is minutes of planning before any data moves. That is why it is off by default for Parquet.

Table formats remove the need. Delta and Iceberg keep the current schema in their metadata, so a reader never opens files to learn it. Schema evolution becomes a metadata operation.

The reasoning

A row in an RDD is a Java object with a pointer to each field, and every field is another object. That costs memory and garbage collection, and the JVM cannot keep it cache-friendly. Tungsten stores a DataFrame row as a compact byte array, `UnsafeRow`, with fields at fixed offsets.

Spark manages that memory itself, with `sun.misc.Unsafe`, rather than leaving it to the garbage collector. Sorting and hashing work on the bytes directly, often comparing prefixes without deserializing anything. That is why a DataFrame shuffle is smaller and faster than the same shuffle on an RDD of tuples.

Whole-stage code generation is the other half. Catalyst fuses a chain of operators into one generated Java function per stage. A filter and a projection become one tight loop instead of a chain of virtual calls. `explain()` shows the fused operators under `WholeStageCodegen`.

The practical lesson is that anything that forces rows out of Tungsten's format is expensive. A Python UDF serializes each row to a Python process and back, which is why native functions and pandas UDFs are preferred.

The reasoning

The configuration page documents `spark.default.parallelism` as the default number of partitions in RDDs returned by transformations like `join`, `reduceByKey` and `parallelize` when not set by the user. Its own default depends on the source: the largest parent RDD's partition count for shuffles, and the total core count across executors, or 2, for `parallelize`.

`spark.sql.shuffle.partitions` applies to the exchanges Catalyst inserts for DataFrame joins and aggregations. It is a flat 200 unless you change it, whatever the cluster size. That is why it is the first thing to tune on a large job, and the first thing AQE coalesces on a small one.

A mixed job can hit both. Read a DataFrame, aggregate, then drop to `df.rdd` and `reduceByKey`: the aggregation used 200 partitions and the RDD step used `spark.default.parallelism`.

The tuning guide's rule for both is 2 to 3 tasks per CPU core as a floor, then size by data volume above that.

The reasoning

By default each task writes one file into each output partition folder it touches. With 200 shuffle partitions and 30 date folders, that is up to 6,000 files. The file count is a product of task count and partition count.

`df.write.option('maxRecordsPerFile', 1000000)` splits a task's output into files of at most a million rows. It bounds the maximum file size but does nothing about tiny files; a task with 10 rows still writes a 10-row file.

To reduce file count, reduce partitions before the write. `coalesce(n)` merges partitions without a shuffle and is the usual choice when shrinking. `repartition(n)` shuffles evenly and is the choice when the data is skewed. `repartition('dt')` before `partitionBy('dt')` sends each date to one task, so each folder gets one file per task that holds that date.

Target 128 MB to 1 GB per file on object storage. Table formats add `OPTIMIZE` or compaction jobs so the write can stay fast and the layout is fixed afterwards.

The reasoning

The streaming guide says to set `checkpointLocation` to a directory in an HDFS-compatible, fault-tolerant file system for sinks where end-to-end fault tolerance can be guaranteed. S3 and ADLS qualify; a local disk on one executor does not.

Inside it, the `offsets` log records which source offsets each batch will process before the batch runs, and the `commits` log records that the batch finished. Together they give replay: a batch that was planned but not committed is run again with the same offsets. For Kafka, Spark tracks offsets here itself and does not commit them to Kafka.

The `state` directory holds the state store, the running aggregations, join buffers and `dropDuplicates` keys, written as versioned snapshots and deltas. With the RocksDB provider, changes are saved to the checkpoint location automatically.

Two rules follow. The checkpoint ties the query to its logic; changing the aggregation or the schema of the state usually needs a new checkpoint. And deleting it means the query starts from `startingOffsets`, which replays or skips data depending on the sink's idempotency.

The reasoning

Append is for results that never change once emitted: a filtered stream, a map, or a windowed aggregation whose windows are closed by a watermark. The guide calls it the default and says it is supported only where rows added to the result table are never going to change.

Complete re-emits the entire result table on every trigger and requires an aggregation, because the engine must hold the whole result. It is practical for small results such as a running top 10, and impossible for unbounded ones.

Update emits only the rows that changed since the last trigger. It is the mode for aggregations without a watermark, where a group's value keeps moving, and for `mapGroupsWithState`. The guide lists that for stream-stream joins only append is supported, and that for other non-aggregate queries complete is not supported because keeping all unaggregated data is infeasible.

The mode interacts with the watermark. In append mode a window is written once, after the watermark passes its end plus the allowed lateness. In update mode the same window is written every trigger it changes.

The reasoning

The Kafka integration guide documents it as a rate limit on the maximum number of offsets processed per trigger interval, proportionally split across topic partitions of different volume. It has no default, so the first batch after downtime is as large as the backlog.

The companion `minOffsetsPerTrigger` makes a batch wait until enough data is available, up to `maxTriggerDelay`, documented at 15 minutes, which trades latency for bigger, cheaper batches.

Spark does not throttle the producer. Backpressure in Structured Streaming is simply bounding how much each batch pulls, then letting the trigger cadence and the batch size set the throughput. If processing is slower than arrival, lag grows and the fix is more parallelism, not a bigger limit.

Parallelism on the Kafka side is one Spark partition per topic partition by default; `minPartitions` can split large Kafka partitions into more Spark tasks. And `failOnDataLoss`, default true, fails the query when the offsets it needs have expired from Kafka.

The reasoning

Each task writes its files under a `_temporary` directory. When a task finishes, its output is committed; when all tasks finish, the job commit moves everything to the destination and writes `_SUCCESS`. The marker comes from Hadoop's `FileOutputCommitter` and can be turned off with `mapreduce.fileoutputcommitter.marksuccessfuljobs`.

On HDFS a rename is one metadata operation. On S3 there is no rename. The store copies and then deletes every object, which the docs call very, very slow. A reader can list the destination while the copy is half done. The docs say the rename-based algorithm is potentially both slow and unreliable.

The documented fixes are committers that do not rename. The S3A committers write files to the final destination as incomplete multipart uploads and complete them only at job commit, configured through `spark.hadoop.fs.s3a.committer.name`, `spark.sql.sources.commitProtocolClass` and `spark.sql.parquet.output.committer.class`. EMR has its own S3-optimized committer, and a manifest committer serves Azure and GCS.

Table formats sidestep the whole problem. Delta, Iceberg and Hudi make the output visible by writing a new metadata version. Readers never see a partial set of files, and `_SUCCESS` becomes unnecessary.

The reasoning

`sc.broadcast(value)` sends a value to each executor once and exposes it through `.value`. The guide says Spark already broadcasts the data a stage's tasks need, so explicit broadcasts are useful when several stages need the same data or when keeping it deserialized matters. Do not modify the value after broadcasting; `unpersist()` releases executor copies and `destroy()` releases it for good.

Broadcast joins use the same mechanism for the small side of a join. That is why the small side must fit in each executor's memory, and why `F.broadcast(df)` is a hint, not a transformation.

An accumulator is created on the driver, added to from tasks, and read on the driver. For updates inside actions, the guide guarantees each task's update is applied once, so a restarted task does not double count. Inside transformations such as `map`, a re-executed task or stage applies its update again, and a lazy transformation that never runs adds nothing.

Two more cautions from the guide. A failed accumulator merge is ignored and the task still succeeds, so a buggy accumulator can be wrong while the job is green. And reading an accumulator before an action runs returns zero.