Apache Spark Architecture

Data Engineering & Modern Data Platforms · Lesson 26 · Big Data Processing and Apache Spark

Lesson Overview

Big Data Fundamentals decides when TechLearningPro's progress landing leaves one process. Hadoop Ecosystem and HDFS decides where the blocks or objects live and how a commit becomes visible. This lesson is the process model in between: which programs run, which of them may die, and how a lazy plan becomes tasks that read those blocks. Apache Spark is the engine the later lessons program in PySpark. The architecture is the part you need when the UI shows a stage that never starts, when a laptop submit dies at midnight, or when an executor disappears and the job continues.

Spark is a distributed data-parallel engine. An application has one driver and many executors. The driver builds a plan. An action on that plan submits a job. The job is cut into stages at shuffle boundaries. Each stage is a set of tasks, one task per partition. A cluster manager — local threads, Spark standalone, YARN, or Kubernetes — gives the application places to run those tasks. The filesystem from lesson 25 gives the tasks bytes. Spark does not replace the contract. A prohibited field still fails the run. A conflicting event_id still fails the run. The checkpoint still moves after a successful commit. What Spark adds is a precise meaning of "the work was split" and a precise meaning of "this piece was retried."

The running system stays the learning-analytics estate. Peak-day volume is still a planning assumption, on the order of fifty million progress events, published as event_date partitions for completion-v3 to read. This lesson uses a tiny local session so you can see a job, a stage, and a partition count without a cluster. Local threads are not a dress rehearsal for YARN. They are a way to make the vocabulary concrete. /learn/python and lessons 4 and 5 are the language on the driver. The executors that run built-in Spark expressions are JVMs. That split is the source of several performance surprises in the next lesson, and it belongs here because it is architecture, not syntax.

Learning Objectives

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

1. Identify the driver, the executors, and the cluster manager, and say which one holds the plan.
2. Define application, job, stage, and task, and mark where a shuffle cuts stages.
3. Explain narrow and wide dependencies in terms of partitions, not in terms of a particular API call alone.
4. Choose cluster deploy mode when the submitter is not allowed to be the driver, and say what client mode binds to the laptop.
5. Describe task retry, executor loss, and driver loss as three different failures, with three different repairs.
6. Read a local explain and a partition count, and state what a two-thread session does not prove.
7. Keep business checkpoints, accumulators, and Spark's own task retries from being confused with one another.

Prerequisites

• Big Data Fundamentals: partitions as units of work, the serial fraction, unsplittable gzip, and the rule against collecting the day to one process.
• Hadoop Ecosystem and HDFS: blocks versus splits, YARN's ResourceManager and ApplicationMaster, and commit visibility. This lesson uses that picture and adds Spark's roles.
• Building Batch Data Pipelines with Python: checkpoint after publish. You should be able to say what a retry is allowed to repeat.
• Data Formats and Serialization: Parquet as the publish format. Spark's tasks read row groups; they do not invent a new file format in this lesson.
• Python 3.11 and, if you run the probe, PySpark 3.5 on Java 17. The probe uses local[2] and a handful of rows. /learn/sql helps when the next lesson writes the same plan as SQL. It is not required to follow the process model.

Detailed Core Concepts

An application is a driver plus its executors

A Spark application is the program you submit for one purpose, such as landing yesterday's progress. It has a driver process. The driver runs your main, holds the SparkSession, builds the logical plan, schedules jobs, and records success or failure. Executors are processes allocated to that application. They run tasks, and they store cached blocks if you asked to cache. One application does not share its executors with another application. Two landings are two applications. If they both target the same event_date, the filesystem will not stop them. The single-writer rule from lesson 14 and lesson 25 stays in the scheduler.

The cluster manager sits under both. In local mode the "cluster" is threads inside one JVM, which is why a laptop can run the probe. In Spark standalone, Spark's own master and workers allocate executors. On YARN, the ResourceManager allocates containers, a NodeManager runs them, and the ApplicationMaster is the driver's home when you choose cluster deploy mode. On Kubernetes, the driver is a pod and executors are pods. Spark's documented cluster managers for the 3.5 line are local, standalone, YARN, and Kubernetes. You pick the one your platform already operates. The landing code should not care which one it is, beyond configuration and the deploy mode. The plan is the same. The failure of the driver is the same shape: the application is dead.

SparkSession is the entry point you construct in PySpark. It owns a SparkContext. You do not need a second context. A session is not a connection pool to a database. It is the handle on the driver. Creating a session per file in a loop, inside tasks, is a misuse. Tasks are not drivers.

Jobs, stages, and tasks

Transformations describe work. Actions run work. filter, select, withColumn, groupBy, and join are transformations. count, collect, take, show, and a write are actions. Until an action runs, Spark has a plan and no tasks. You can explain the plan for free relative to the data volume, because explain does not read the progress file. This is the operational meaning of laziness. It is also a trap: a method that looks like a harmless print may be an action. count to "see if the file is empty" reads the file.

An action submits a job. The driver's DAG scheduler cuts the job into stages. A stage boundary is a shuffle, which is required when a partition of the output depends on many partitions of the input. Reading, filtering, and projecting can stay inside a partition: each output partition comes from one input partition. That is a narrow dependency, and those operations pipeline into one stage. Grouping by event_id to compare canonical hashes needs all copies of that id together. Those copies may sit on any input slice. The exchange is a wide dependency and a new stage. The first stage writes shuffle files. The second stage's tasks fetch the files they need. A job that validates lines and then checks collisions is at least two stages if the check is a grouping. A job that only filters and writes, with no grouping, may be one stage plus the commit.

A task is the work of one stage on one partition. One hundred input slices are one hundred tasks in the read stage, if nothing coalesces them. Tasks are the unit of retry. Stages are the unit of the barrier: the next stage waits until the shuffle outputs it needs are available. The job is the unit of the action. The application is the unit of the driver. When someone says "the task failed" and someone else says "the job failed," they are pointing at different scopes. This lesson uses the four words consistently. The UI uses them the same way.

Where the bytes go

A task on an executor reads its split from HDFS or from the object store. It does not ask the driver to proxy the file. The driver tells the task which split to read. That is why the NameNode or the object store must be reachable from the executors, not only from the submitter's laptop. Client deploy mode makes this easy to get wrong: the driver runs on the laptop, where hdfs dfs -ls worked because the laptop has a keytab, and the executors in the cluster fail because their network path or their token is different. Cluster deploy mode runs the driver next to the executors. The laptop only submits.

Shuffle bytes are a second network. Map tasks write shuffle blocks, usually to local disk on the executor. Reduce tasks pull those blocks. If an executor is lost after writing shuffle output, the output may be lost with it, and the map tasks that produced it are recomputed. An external shuffle service, used especially with dynamic allocation on YARN, keeps shuffle files on the node so executors can come and go. Turning on dynamic allocation without the service the documentation requires for that cluster manager deletes shuffle data out from under the next stage. The symptom is a stage that reruns map tasks you thought were finished. The repair is the documented service, or leaving dynamic allocation off until you have it.

Cache is a third place bytes sit. cache or persist marks a dataset to be kept on executors after the first action that computes it. The memory manager carves the executor heap into reserved memory, a user region, and a unified region shared by execution (sorts, shuffles, joins) and storage (that cache). The defaults in Spark 3.5 — spark.memory.fraction at 0.6 and a storage fraction inside it — mean cache and shuffle compete. Caching an 18 GB day on a small executor pool spills or evicts, and the next action recomputes. Cache a dataset when two actions in the same application reuse it and you have measured the reuse. Do not cache the landing "so it is in memory" as a default. The landing's useful reuse, if any, is a small aggregation, not the raw event body.

Deploy mode and cluster managers

In client deploy mode the driver runs in the spark-submit process. Logs appear in that terminal. If the process exits, the application exits, and executors are removed. A laptop sleep, a closed SSH session, or a deploy host that reboots is a driver failure. Client mode is appropriate for a local probe and for an interactive session you are watching. It is the wrong mode for the 01:00 UTC landing.

In cluster deploy mode the driver runs inside the cluster: as the YARN ApplicationMaster, or as the Kubernetes driver pod. spark-submit uploads the application and returns once the application is accepted, depending on flags. The cluster keeps the driver. Logs go to the cluster's log system, not to a terminal you closed. This is the mode for the scheduled landing. The scheduler still has to notice a failed application and decide whether to rerun it. Spark does not restart a dead driver by itself. The rerun is safe only because the commit and the checkpoint match lesson 14.

Resource settings travel with the submit. spark.executor.memory is the executor JVM heap. spark.executor.memoryOverhead is extra native memory, important when off-heap shuffle and Parquet readers allocate outside the Java heap. spark.executor.cores is how many tasks an executor runs at once. spark.driver.memory matters when the driver collects results or holds a large plan. It should not need to hold the day. spark.sql.shuffle.partitions defaults to 200 for SQL shuffles. Two hundred tasks for a collision check on a small fixture is noise. Two hundred tasks for a multi-terabyte backfill may be too few, and each task's shuffle record becomes large. Set it from data size in the job configuration, and revisit it when the peak day grows. Lesson 30 is the deeper shuffle tuning. Leaving 200 unexamined on both the fixture and the quarter restatement is how a plan that looked fine locally thrashes in production, or the reverse.

Failure is three mechanisms

Task failure is the common case. The executor throws, the driver schedules the task again, up to spark.task.maxFailures (four by default). A bad record that always throws will burn the retries and fail the stage. That is what you want for a contract breach you coded as an exception, and it is what you do not want for a single malformed line you meant to quarantine. Quarantine is a written row, not a thrown exception, or one bad line fails the morning after four identical stack traces. A transient disk error on one node often succeeds on the retry because the retry lands elsewhere.

Executor loss removes a process. Its tasks are rescheduled. Its cached blocks are gone and will be recomputed from lineage if an action still needs them. Lineage is the record of how to recompute a partition from its inputs. Lesson 29 opens that record. Here the operational sentence is: input files must still be in place for the recompute to work. If the landing deletes or renames inputs halfway through, retries have nothing to read. Keep the closed input stable until the checkpoint moves.

Driver loss removes the plan, the stage progress, and any in-memory accumulator state. Executors go away. The cluster manager marks the application failed. A rerun starts from the beginning of the application, not from the stage that was in flight, unless you built that yourself. The landing does not need to resume a stage. It needs to replace the date partition idempotently and write the checkpoint once. Partial task outputs belong to the committer's attempt directories, which lesson 25 kept invisible to readers. Driver death before commit leaves the previous publish. Driver death after commit and before checkpoint causes a full rerun that must replace, not append.

Speculation (spark.speculation) launches a duplicate of a slow task. The committer keeps one winner. Speculation is legal when tasks write through the committer and do not update external systems themselves. It is not legal as a way to run two drivers.

Accumulators and broadcasts are driver contracts

A broadcast variable is a read-only value the driver ships to executors, replicated so tasks do not each pull a large copy from the driver. A small map of lesson_id to course id can be a broadcast when a later lesson joins it. The progress file cannot. Broadcasting tens of gigabytes exhausts executor memory and the driver's network. If the dimension is large, it is a join, and lesson 30 treats it as a join.

An accumulator lets tasks add to a value the driver reads, for counts and diagnostics. Spark's own documentation limits the exactness guarantee: updates inside actions are applied once per successful task, and updates inside transformations can be applied more than once if a task or stage reruns. A collision count computed by incrementing an accumulator inside a transformation can over-count after a retry and fail a clean file, or, if you used it as a success signal, lie. The collision rule in this course is a grouping that writes a result or fails the job from that result. Counters in the UI are for operators. The checkpoint does not read an accumulator as its source of truth.

What the UI is for

The driver serves a web UI on port 4040 while the application runs, moving to the next port if 4040 is taken. It shows jobs, stages, tasks, shuffle read and write, and failures. Enable the event log (spark.eventLog.enabled) and a history server if operators must look at a landing after the driver has exited. Without the event log, a dead application takes its UI with it. The history server is how the 05:00 UTC incident gets a stage timeline instead of a verbal recollection.

Look at the number of tasks, the duration spread, the bytes shuffled, and the failure reason. A single task running for the whole input is the unsplittable file from lesson 24. Tens of thousands of tasks that finish in tens of milliseconds are the small-file layout from lesson 25. A shuffle measured in tens of gigabytes for a collision check that should move only ids and hashes means the plan shuffled payloads. The UI confirms the architecture. It does not replace the contract tests.

Python on this architecture

In PySpark the driver code you write is Python. It communicates with the JVM driver to construct the plan. Built-in column expressions run as JVM tasks on executors. A Python UDF is different: each executor starts a Python worker, and rows cross a socket between the JVM and that worker. That boundary is architecture. It serializes data, it breaks the JVM's ability to compile the whole stage into one tight loop, and it puts your process model on two runtimes. The next lesson uses built-in functions for the landing for that reason. This lesson's rule is: if a step must be Python, it should be a deliberate boundary with a measured cost, not the default way to parse a timestamp.

The driver Python process is also a place to do small work after an action: read a count that came back, decide the run fails, write the checkpoint file with the same replace-then-record order as lesson 14. That Python is not distributed. Keep it small. A checkpoint write of one JSON object is appropriate. A loop over fifty million rows in the driver is the collect anti-pattern.

Architecture and Design Diagrams

The control path and the data path are different. The driver and the NameNode are control. Executors and DataNodes are data. Mixing them is how a laptop becomes a bottleneck.

mermaidCopy
flowchart TD
    submit[spark-submit or scheduler]
    cm[Cluster manager]
    driver[Driver: SparkSession and DAG scheduler]
    ex1[Executor tasks]
    ex2[Executor tasks]
    store[(HDFS or object store)]

    submit --> cm
    cm --> driver
    cm --> ex1
    cm --> ex2
    driver -->|task descriptions| ex1
    driver -->|task descriptions| ex2
    ex1 -->|read and write splits| store
    ex2 -->|read and write splits| store
    ex1 -.->|shuffle blocks| ex2

An action becomes a job. A wide dependency cuts a stage. Tasks match partitions.

mermaidCopy
flowchart LR
    action[Action such as a write or count]
    job[Job]
    stage1[Stage: read, project, validate locally]
    stage2[Stage: group by event_id]
    tasks1[One task per input partition]
    tasks2[One task per shuffle partition]

    action --> job --> stage1 --> tasks1
    stage1 -->|shuffle ids and hashes| stage2 --> tasks2

Driver death and task death do not leave the same artifact. The checkpoint is outside both of them, written by the driver only after commit.

mermaidCopy
flowchart TD
    run[Application running]
    taskdie[Task fails]
    execdie[Executor lost]
    drv[Driver lost]
    commit[Partition committed]
    ckpt[Checkpoint written]
    retryTask[Task retried up to maxFailures]
    recompute[Lost shuffle or cache recomputed]
    rerun[Scheduler reruns the whole application]

    run --> taskdie --> retryTask
    run --> execdie --> recompute
    run --> drv --> rerun
    run --> commit --> ckpt
    drv -.->|if before commit| rerun
    commit -.->|if driver dies before checkpoint| rerun

Technical Explanation

Lazy plans and the moment a job exists

The probe in the walkthrough builds a DataFrame, filters it, and groups it. None of that reads data until an action. explain prints the physical plan the driver has constructed: scans, projections, exchanges. Exchanges are shuffles. When you call an action, the driver submits a job, the DAG scheduler materializes stages, and the task scheduler places tasks on executors that the cluster manager has allocated. If the cluster manager has allocated nothing yet, the first job waits in a pending state that looks like a slow function call. On YARN that wait is a queue. The architecture is working. The queue is the thing to read.

Whole-stage code generation, part of Spark's execution engine since Spark 2, compiles chains of operators in a stage into bytecode so a filter and a project do not pay a virtual call per row. You see one stage in the UI, not a box per operator. Catalyst, the optimizer, may reorder filters and push a date predicate into a Parquet read. Lesson 27 reads the optimized plan. The architecture claim here is only that the stage you scheduled is not a literal transcript of the Python line order, and that this is a feature when the expressions are the engine's own.

Partition counts you will actually meet

The input partition count for files follows splits: block alignment on HDFS, spark.sql.files.maxPartitionBytes, and whether the codec splits. The shuffle partition count for a SQL grouping follows spark.sql.shuffle.partitions unless adaptive query execution changes it. Adaptive query execution, spark.sql.adaptive.enabled, can coalesce shuffle partitions after it sees their sizes. It is on by default in many 3.5 deployments and is still a setting you should look up for the build you run. It does not fix a skewed key by magic in every case, and it does not change the directory partition event_date. Three different "partitions" remain in the vocabulary: input splits, shuffle partitions, and output directory partitions. Say which one you mean in a design review.

repartition(n) is a shuffle to n partitions. coalesce(n) reduces partitions and can avoid a full shuffle when you are only combining adjacent ones. Using coalesce(1) before a heavy parse forces the parse into one task. Using repartition(1) before the write of a multi-gigabyte day creates one huge file and one long task. The write-side file count is a deliberate repartition to a moderate n, then partitionBy("event_date"), implemented in lesson 28. The architecture reason is task size and commit size, not a preference for round numbers.

Memory failures look like retries

An executor out-of-memory kill shows up as a lost executor and a task retry. If the task's working set is larger than any executor, every retry dies the same way and the stage fails after spark.task.maxFailures. Adding executors does not help if each one is too small for the hot key. Making executors larger helps only if the skew fits in one of them and the rest of the cluster can spare the memory. The landing's collision record is small on purpose so this situation means a producer bug or a plan that shuffled the full payload. The UI's shuffle write size tells you which.

Driver out-of-memory is different. The driver was asked to hold a result, a broadcast, or a huge plan. collect of the day is the usual cause. Increase driver memory only for a result you have proved is small, such as a few aggregation rows. A growing driver heap during collect is a bug in the plan.

Tokens, principals, and the trust boundary

On a Kerberos HDFS cluster, the driver obtains delegation tokens and distributes them so executors can read the prefix. The submitter's identity is the identity the NameNode sees, subject to proxy rules. Run the landing as the landing principal from lesson 25, not as a personal account that happens to have a home directory of progress samples. On an object store, the credentials belong to the execution environment — a role the pods or containers assume — and not in a Python string in the repository. Spark configuration can hold secrets if you put them there. Do not. The event log will outlive the job and is a bad place for a payload or a key. spark.eventLog records the plan and metrics. It should not record rows. It will record your SQL text. Do not embed a learner id list in a query string you submit.

Network encryption (spark.network.crypto.enabled) and local IO encryption (spark.io.encryption.enabled) default off in Spark 3.5. Shuffle files and cached spills of progress data are learner activity. Enable the settings your release documents if executors run outside a disk trust boundary you already accept. Replication on HDFS does not cover executor-local shuffle disks.

Local mode is a test of the plan

local[2] runs two task threads and the driver in one JVM. It will show you a shuffle exchange in explain. It will run both stages. It will not show you a NameNode, a queue, a token, a container kill, or a network. Tests of contract logic belong here, on fixtures. Tests of commit visibility belong on the target filesystem, even if the fixture is small. Tests of the morning bound belong on a measured slice of real volume in an environment that matches production closely enough that the rate means something. The architecture supports all three. One green unit test covers only the first.

Practical Implementation or Design Walkthrough

Problem

Show, on a laptop, that a transformation builds no job, that a grouping introduces an exchange, and that an action creates work. Map each visible fact to the production landing without pretending the laptop is the cluster. Pin the session to PySpark 3.5 APIs.

Session

pythonCopy
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

PROGRESS_EVENT_SCHEMA = StructType(
    [
        StructField("event_id", StringType(), nullable=False),
        StructField("learner_id", StringType(), nullable=False),
        StructField("lesson_id", StringType(), nullable=False),
        StructField("event_type", StringType(), nullable=False),
        StructField("attempt_number", IntegerType(), nullable=False),
        StructField("event_time", StringType(), nullable=False),
        StructField("checkpoint", IntegerType(), nullable=True),
    ]
)


def architecture_session() -> SparkSession:
    return (
        SparkSession.builder.master("local[2]")
        .appName("tlp-architecture-probe")
        .config("spark.ui.enabled", "false")
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.sql.adaptive.enabled", "false")
        .config("spark.sql.session.timeZone", "UTC")
        .getOrCreate()
    )

Adaptive execution is disabled in the probe so the exchange in the plan stays where you can see it. Production may leave it enabled. The probe would then be a bad picture of production partition counts and a good picture of the logical shuffle. Say which configuration you are looking at before you compare screenshots.

Plan, then action

pythonCopy
def collision_plan(spark: SparkSession, path: str):
    events = spark.read.schema(PROGRESS_EVENT_SCHEMA).json(path)
    projected = events.select("event_id", "event_type", "event_time")
    # Narrow so far: one input partition produces one output partition.
    counted = projected.groupBy("event_id").agg(F.count("*").alias("copies"))
    return counted


def demonstrate(spark: SparkSession, path: str) -> int:
    counted = collision_plan(spark, path)
    counted.explain(mode="formatted")
    # The explain above does not submit a job. The count below does.
    spark.sparkContext.setJobDescription("progress-architecture-probe-count")
    return counted.count()

explain(mode="formatted") is the Spark 3 plan printer. In the formatted plan, look for a FileScan or JSON scan, a projection, and an Exchange before the aggregate. The exchange is the stage boundary. count is the action that submits the job. setJobDescription labels that job in the UI when the UI is enabled, which is how a landing with several actions stays readable at 05:00 UTC. The probe turns the UI off so tests do not bind a port. Production turns the event log on.

The return value is a single integer, safe to bring to the driver. Replacing count with collect on projected would pull every event to the driver. On the fixture it works. On the peak day it is the out-of-memory incident from lesson 24, now with a cluster attached.

What you configure when you leave the laptop

A scheduled submit uses cluster deploy mode and the landing principal. The values are an example shape, not a capacity plan. Memory and executor counts come from the measurement memo.

textCopy
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name tlp-progress-landing \
  --executor-memory 4g \
  --executor-cores 2 \
  --num-executors 8 \
  --driver-memory 2g \
  --conf spark.sql.shuffle.partitions=64 \
  --conf spark.eventLog.enabled=true \
  --conf spark.task.maxFailures=4 \
  progress_landing.py --input hdfs://nameservice1/data/tlp/inbox/progress-2026-03-02.jsonl

--master yarn and --deploy-mode cluster are documented spark-submit flags. Kubernetes swaps the master URL for k8s:// and adds an image and a namespace. The script's arguments carry the input path. The code does not contain the date. A retry of Tuesday changes the argument, not the Python. Two submits of the same date are still two applications. The scheduler, not spark-submit, enforces one at a time.

progress_landing.py does not exist yet as the production job. Lesson 28 writes it. This lesson's probe is the architectural slice: schema, narrow projection, wide aggregation, action, job description. If demonstrate is called twice on the same plan without cache, Spark computes twice. That is laziness plus the absence of reuse, and it is what you want to see once so you do not put two writes on one plan and assume the read happens once without checking the UI.

Reading the result against the contract

The probe counts copies. It does not yet compare canonical hashes, quarantine lines, or write Parquet. A copies value above one is a duplicate id, which may be a legal retry of the same payload. The probe cannot tell. Lesson 27 adds the hash. Using this count as a refusal would fail clean retries. Architecture probes that quietly become production checks are how contract bugs ship. Keep the probe in a test module.

Enterprise Real-World Scenarios

The laptop that was the driver

An engineer proved the landing on a YARN cluster in client deploy mode so the logs streamed to a terminal during a late test. The scheduler was then pointed at the same command without --deploy-mode cluster. At 01:10 UTC the submit host, a shared interactive node, was rebooted for a patch. The driver died with the node. Executors were killed. The date partition had not committed. The checkpoint had not moved. The previous day's publish remained, so completion-v3 was stale and not wrong. The incident channel described it as "Spark lost the data." The data was on HDFS, unharmed. The application was dead because the driver had been placed on a machine nobody considered part of the application.

The repair was cluster deploy mode, a driver memory setting written down, and the event log so the next failure could be read after the driver was gone. The scheduler's alert was attached to the application id, not to an SSH session. A rerun after the reboot replaced nothing, because nothing had committed, and then committed once. The postmortem listed three failure scopes — task, executor, driver — and which one the reboot was. It also noted that client mode remains correct for the local[2] tests in CI. The production submit path rejects a missing --deploy-mode cluster in review. The code did not change. The place the driver runs changed.

Two hundred partitions and fifty thousand tasks

A backfill used the default spark.sql.shuffle.partitions of 200 on a quarter of progress, and it also read the partner small-file tree from lesson 25 because the compact job had not finished. The UI showed a read stage with tens of thousands of tasks, most of them shorter than the time spent scheduling them, and a shuffle stage with 200 tasks, some of them large because a few event_id values were retry storms. The driver CPU was high. The executors were mostly idle in the read stage and then uneven in the shuffle. Adaptive execution was off in that older configuration, so nobody coalesced the read.

The architectural repair was to compact first, so the read stage had hundreds of tasks that each read a real slice, and to set shuffle partitions from the volume of the collision record rather than from the default. A hot key still showed up as one long task. That task was a producer bug, fixed by failing the file when a single id exceeded a configured copy cap, which is a contract decision implemented as a filter on the aggregation result. Adding executors before the compact would have added more idle capacity and a larger scheduling load. The UI's task-duration histogram was the evidence. The memo from lesson 24 was updated with the observed task counts so the next backfill starts from the compact tree.

Security noted that the event log from the failed configuration contained the SQL text of a debugging query an engineer had run in the same application, including a learner_id they had pasted into a filter. The repair for that was procedural: ad hoc queries on raw progress run as a separate application with a short retention on the event log, and production job text stays on contract columns without literal learner ids. The architecture allows many applications. It does not classify the event log for you.

Best Practices

• Run the scheduled landing in cluster deploy mode. Use client mode for interactive debugging and local[2] for plan tests.
• Treat application, job, stage, and task as different scopes in every incident note.
• Let transformations stay lazy. Know which calls are actions. Put setJobDescription on each action a human will see in the UI.
• Keep the driver small: configurations, the plan, counts, the fail-or-continue decision, the checkpoint write. Do not collect events.
• Set spark.sql.shuffle.partitions from the shuffle's data size. Revisit it when the peak day changes.
• Cache only a reused dataset you have measured. Expect cache and shuffle to share executor memory.
• Leave inputs immutable until the checkpoint moves, so task and executor retries can recompute.
• Enable the event log and the history server for any application that runs unattended.
• Quarantine bad lines by writing them. Reserve task failure for breaches that should fail the stage.
• Do not use a transformation accumulator as the collision source of truth.
• Enable the documented network and local IO encryption when shuffle disks leave the raw-zone trust boundary.
• One application writer per event_date. The cluster manager will run two if you ask.

Common Mistakes and Anti-Patterns

Client mode in the scheduler. The driver dies when the submit host dies, and people describe it as a Spark outage. The repair is cluster deploy mode and an event log. The previous publish stays if the commit had not happened. That part is success. The surprise is the failure.

collect of the day. The driver heap becomes the landing. Retries of the action make it worse. The repair is an aggregation that returns a scalar or a few rows, with the heavy work left on executors.

One giant coalesce. coalesce(1) before the parse or the write turns the application back into the single process lesson 24 measured, plus scheduling overhead. The repair is a partition count chosen for slice size, reviewed in the UI.

Dynamic allocation without the shuffle service. Executors disappear, shuffle files disappear with them, and map stages rerun until the job misses the bound. The repair is the external shuffle service or shuffle tracking your cluster manager documents, or static executors for a short daily job that does not need to scale down mid-query.

Caching the raw day by default. Execution memory and storage memory contend, spills increase, and the job is slower than the uncached plan. The repair is to cache only a small reused aggregate, and to unpersist it. Measure with the UI's storage tab, not with hope.

Failing the task on a quarantinable line. Four identical exceptions, then a failed stage, then a stale mart, because one truncated line was thrown rather than written aside. The repair is a permissive parse that writes rejects, and an exception only for file-level breaches you want to abort on.

Sharing a SparkContext across unrelated submissions inside one long-lived process without isolation. A notebook server that keeps a session up and runs two dates at once will violate the single-writer rule inside what looks like one application. The repair is one application per landing run, started by the scheduler, and no shared mutable path.

Reading the UI from a sample and provisioning production from it. local[2] with shuffle partitions at 4 shows four tasks. Production at 200, or at 50,000 input files, is a different application shape. The repair is to look at the production event log after the first full day and adjust, before enrollment week freezes the configuration.

Troubleshooting and Design Review

Symptom: The application sits with no tasks, and executor count is zero.

Likely cause: the cluster manager has not allocated resources. The queue is full, the principal cannot request containers, or the Kubernetes pods are pending.

Investigation: the cluster manager's queue or pod status, not the Spark plan. explain will still look healthy because it does not need executors.

Corrective action: fix the allocation. Do not rewrite the transformation. Once executors appear, confirm tasks start. A pending application is not a slow shuffle.

Symptom: Tasks start, one runs, the others stay pending, and the input is one file.

Likely cause: the file is unsplittable, or it is smaller than the split target, so there is one partition.

Investigation: input partition count, codec, file size, spark.sql.files.maxPartitionBytes.

Corrective action: split the input as lesson 24 described. More executors remain idle until then.

Symptom: A stage fails after four task attempts with the same exception on a record.

Likely cause: deterministic bad input, or a bug in a UDF, or an executor that is too small so every attempt is killed with the same out-of-memory error.

Investigation: the exception class. An OutOfMemoryError is capacity or skew. A parse exception you raised is the contract. Four identical stack traces mean retry will not heal it.

Corrective action: for a contract breach, fail the run and keep the previous publish. For memory, shrink the record or the skew before you raise maxFailures. Raising retries hides a deterministic bug behind a longer outage.

Symptom: Map tasks run twice, and nobody resubmitted the application.

Likely cause: an executor was lost and its shuffle output went with it, so the scheduler recomputed the map stage. Dynamic allocation without a shuffle service does this on purpose, unfortunately.

Investigation: executor lost events in the UI, and whether dynamic allocation and the shuffle service match the docs for your cluster.

Corrective action: stabilize the executors or enable the shuffle service. Confirm inputs were not deleted. The recompute is lineage working. It is expensive if it happens every stage.

Symptom: The history server has no record of last night's failure.

Likely cause: the event log is disabled, or it is on a local disk of a driver that died in client mode, or the history server cannot read the log directory.

Investigation: spark.eventLog.enabled and spark.eventLog.dir on the submit, and the history server's directory.

Corrective action: write the event log to the same durable filesystem class as the landing, with a retention period. Do not put learner payloads in queries that get logged as text.

Symptom: Two successful applications in the same hour, and the date directory has mixed part files.

Likely cause: two drivers, two commits, no single-writer lock.

Investigation: two application ids, two checkpoint writes or a checkpoint that matches only one hash.

Corrective action: stop the scheduler's overlap, rewrite the date from the winning input, and keep cluster mode from being blamed. The cluster did what it was asked. It ran both.

A design review asks: Where does the driver run? What is an application versus a job in this landing? Where are the shuffles, and what record crosses them? What is the input partition count on a peak day, and the shuffle partition count? What happens on task failure, executor loss, and driver loss? Where is the event log after the driver exits? What does the driver collect? Who prevents two applications from committing one date? The review can be done with a diagram and a failed-job history page. It does not require a new engine.

Interview Questions and Answers

Q1. What runs in the driver, and what runs in an executor?

The driver runs my program's main path, the SparkSession, planning, scheduling, and the final decision to fail or checkpoint. Executors run tasks on partitions and hold cached blocks. Built-in Spark expressions run in the executor JVM. I do not read the progress file through the driver. I do not construct a SparkSession inside a task. If the driver needs a result, it is an aggregation. For TechLearningPro that result is counts, a collision flag, and then a checkpoint write of a small record.

Q2. What is the difference between an application, a job, a stage, and a task?

An application is one driver and its executors, for example one landing run. A job is one action. A stage is a slice of that job between shuffles. A task is one partition of one stage, and it is the unit of retry. A landing that filters, groups by event_id, and writes may be one application, two or three jobs if I count separate actions, two stages around the group, and a task count tied to splits and to spark.sql.shuffle.partitions. I use those words in incident notes so "it failed" names a scope.

Q3. Why do transformations not read the file?

They add to a plan. The file is read when an action forces a job. explain shows exchanges and scans without that read, which is how I review a landing without scanning 18 GB. The trap is an action hiding in a debug print. count and collect and writes are actions. I structure the code so the production actions are the collision check and the commit, both labeled with a job description.

Q4. What makes a shuffle a stage boundary?

A wide dependency: an output partition needs data from many input partitions. Grouping by event_id is wide because duplicates can be anywhere in the file. Filtering a bad event_type is narrow because each row decides alone. Spark pipelines narrow steps into one stage and cuts a stage at the exchange. The next stage starts when the shuffle data it needs is available. That barrier is why one slow map task holds the collision check.

Q5. Why is client deploy mode the wrong production default?

The driver runs in the submitter. If that machine sleeps, reboots, or loses the session, the application dies and executors are removed. Cluster deploy mode places the driver in the cluster — the YARN ApplicationMaster or the Kubernetes driver pod — and the submitter only submits. I still need the scheduler to notice a failed driver and rerun. Spark will not resurrect the driver. The rerun is safe because commit and checkpoint are idempotent, not because deploy mode is a transaction.

Q6. A task fails four times with an OutOfMemoryError. What do you change first?

I do not raise spark.task.maxFailures. Four identical memory failures mean the task's working set does not fit the executor. I look at shuffle bytes and key skew. For the collision check I want the shuffled record to be an id and a hash. If a single id has a huge number of copies, I fail the run as a producer incident. If the executor heap is simply below the documented need of a healthy task, I resize from the measurement. More executors that are each too small repeat the failure in parallel.

Q7. What does executor loss do to cached data and shuffle files?

Cached blocks on that executor are lost and recomputed from lineage if needed. Shuffle files on its local disk are lost unless an external shuffle service retained them, and the map tasks that produced them are rerun. Inputs must still be where the lineage says they are. I do not delete the inbox file at task start. I delete or expire it after the checkpoint, under the retention rule.

Q8. How do accumulators mislead a landing?

Updates performed in transformations can be reapplied when a task or stage reruns, so a collision counter can move without a second real conflict. I use accumulators as rough operator metrics if at all. The refusal decision comes from a grouped result: more than one canonical hash per event_id. The checkpoint records counts taken from that result after a successful action, not a running tally inside map.

Q9. What does local[2] prove?

It proves the plan runs, the schema is accepted, narrow and wide steps do what I expect on a fixture, and my assertions on a handful of rows hold. It does not prove tokens, queues, commit semantics on the object store, executor loss, or the morning bound. I keep contract tests on local[2] with the UI off and shuffle partitions small. I keep a separate note for the filesystem commit test and the volume test so a green CI job is not described as an enrollment-week proof.

Q10. Where should spark.sql.shuffle.partitions be set, and what is wrong with the default?

It should be set in the application configuration from the size of the data that will be shuffled, and reviewed when volume changes. The default of 200 is a constant. On a tiny probe it creates too many tiny tasks if I leave it, which is why the probe sets 4. On a large backfill, 200 tasks can make each reducer hold a very large spill, and on a small-file read it does not even control the read-stage task count. Input splits and shuffle partitions are different. I set both on purpose. Adaptive execution may coalesce shuffle partitions later. I still choose an initial number I can defend.

Q11. Why is a Python UDF an architecture decision?

Built-in expressions stay in the executor JVM and can be compiled into the stage. A Python UDF runs in a Python worker process beside the executor, with serialization across that boundary. I use it when the logic cannot be expressed in the engine, and I measure it. Timestamp normalization and the contract fields of a progress event can be expressed with built-in functions, so the landing does not pay the boundary per line. The driver remains Python either way. That is the right place for Python: orchestration of the plan, not per-row work.

Q12. What should an operator see after a driver dies at 02:00 UTC?

A failed application in the cluster manager, an event log in the history server if we enabled it, an unchanged checkpoint, and a published date partition that is either the previous complete version or a committed new version. They should not see a partial part-file list as the dataset readers scan, and they should not need the engineer's laptop. If the commit finished and the checkpoint did not, the rerun replaces that date and then writes the checkpoint. I confirm that sequence in a drill. I do not reconstruct the night from memory of the 4040 UI, because that UI left with the driver.

Hands-On Exercises

Exercise: Trace one action from plan to failure scope

Problem statement. Use a small progress fixture and a local Spark session to show a narrow step, a shuffle, and an action. Write the production failure story next to it without claiming the local session is the cluster.

Tasks.

1. Build a session at local[2] with an explicit shuffle partition count and UTC timezone. Use the progress schema.
2. Define a projection and a groupBy on event_id. Call explain before any action and identify the exchange.
3. Run one action that returns a scalar. Set a job description. Record that a second action without cache computes again.
4. State the production deploy mode and where the driver runs. Contrast it with this exercise.
5. Write what task failure, executor loss, and driver loss do to the published date and the checkpoint.
6. Say what collect on the event projection would do to the driver at peak-day scale, using the planning assumption as a labeled estimate.
7. Name one UI signal for unsplittable input and one for small files.
8. Apply the six-part lens to cluster versus client deploy mode, including who can read shuffle spills.

Expected deliverables. The probe script, an explain note, and the failure-scope write-up.

Evaluation rubric.

You pass when a reviewer can point to the exchange in your plan note and then say, without prompting, what remains published if the driver dies before commit.

Lesson Summary

A Spark application is a driver and its executors, placed by a cluster manager. Transformations build a plan. An action submits a job, which the driver cuts into stages at shuffles and into tasks at partitions. Narrow work stays in a stage. Grouping by event_id does not. Task retries, executor loss, and driver death are different failures. Only the last one ends the application, and the scheduler must rerun it. Cluster deploy mode keeps the driver off the laptop. The checkpoint still moves only after commit, and the driver is the place that writes it, after a small result comes back. Local mode proves the plan on a fixture. The UI and the event log prove what the cluster did with the real file.

Spark DataFrames and Spark SQL is the next lesson. It uses this process model to express the progress landing as a schema, a set of transformations, and a few deliberate actions.

Further Reading

• Spark cluster overview — driver, executor, cluster manager, and deploy mode. The vocabulary in this lesson tracks that page.
• Spark submitting applications — spark-submit, --deploy-mode, and executor sizing flags.
• Spark configuration — spark.sql.shuffle.partitions, spark.task.maxFailures, memory fractions, event log, and speculation. Confirm defaults on the build you run.
• Spark security — network crypto, local IO encryption, and how that meets the raw-zone boundary.
• Spark monitoring — the 4040 UI and the history server.
• YARN overview — the cluster manager most likely to appear under a first production submit.
• Continue this course: Spark DataFrames and Spark SQL. The storage layout remains Hadoop Ecosystem and HDFS.