Spark, simulated

The diagram. The driver (with what its DAG scheduler and task scheduler are doing, the map outputs it tracks and the cached blocks it knows about), the cluster manager (YARN, Kubernetes or standalone) and one card per executor. Each executor shows one row per core — the running task, coloured by its stage, filling as it progresses (hatched when it spilled) — and a memory bar: reserved 300 MiB, user memory, the unified region with storage (cached blocks, broadcasts) growing from the left and execution (tasks) from the right, a tick at the storage region that execution cannot evict, and the container overhead outside the heap. Packets: task launches (orange), status updates (green, red when a task failed), shuffle fetches between executors (blue, as long as the transfer takes), broadcasts and results going to the driver (pink), and the cluster manager’s requests, container starts, registrations and lost-executor reports (violet). Tap the driver, the cluster manager or an executor to inspect it, kill or slow an executor, or add one.

Stats. Jobs, tasks, executors and slots, input read, shuffle written and fetched remotely, spill (memory and disk, as the Spark UI reports them), failed, recomputed and speculative tasks, storage used.

Task timeline. A Gantt chart with one lane per executor core: every task attempt of the latest job, coloured by stage; stragglers (more than twice the stage median) outlined red, spilled tasks hatched, failures crossed, speculative copies dashed. Hover for a task’s input, shuffle read, rows and spill.

Stages (DAG). The stages of the latest job with their dependencies (shuffles solid, broadcasts dashed), progress, and — tap one — the summary metrics of its finished tasks (min, 25th percentile, median, 75th percentile, max of duration, input, records and spill), as on the Spark UI’s stage page.

Program. The PySpark (or SQL) program. Apply parses and plans it; Run ▶ runs it with its own action or the one you choose. The output shows the job’s result — row counts are modelled from the table statistics; the rows shown are computed on a 400-row sample of each table — or the explain() text. Examples & tables lists the sample tables, the supported methods and functions, and some programs to start from.

Plans. The parsed (unresolved) plan, the analysed plan with expression ids and types, the optimised plan, the optimizer rules that changed it (each with the plan before and after), the physical plan as df.explain() prints it (with the join-selection reasoning under it), the final adaptive plan after the job ran (query stages, AQEShuffleRead, join changes, whole-stage-codegen ids), and the formatted and cost explain modes.

Do something. Run the program or count()/show()/write, kill or slow any executor, add one, kill one during the next reduce stage, and toggle AQE, skew join handling, speculation, dynamic allocation and the sort-merge preference.

Event log. Key events, or everything, in the style of Spark’s own driver log (DAGScheduler, TaskSetManager, ShufflePartitionsUtil, MemoryStore, ExecutorAllocationManager…).

Settings rebuild the application: cluster manager, executors, cores, executor and driver memory, memory overhead and off-heap, spark.memory.fraction and storageFraction, shuffle partitions, the broadcast threshold, AQE options, file split size, skew in the data, native memory per task, dynamic allocation and speculation. Each setting shows Spark’s default.

Sizing — how many executors, how many cores and how much memory each, from the nodes you have (leave a core and some memory per node for the OS and daemons; about five cores per executor), with the container’s memory split into reserved, user, storage and execution regions and overhead, task slots, waves, and partition counts for a data size; it writes the spark-submit line. Presets include the worked example from Cloudera’s "How-to: Tune Your Apache Spark Jobs (Part 2)" (6 nodes × 16 cores × 64 GB → 17 executors × 5 cores × 19 GB).

EXPLAIN reader — paste the output of df.explain(), df.explain("formatted") or EXPLAIN FORMATTED … (Spark 3.x and 4.x, including the final plan of an adaptive query) and get it as a tree: what every operator does (Exchange, BroadcastExchange, SortMergeJoin, BroadcastHashJoin, ShuffledHashJoin, HashAggregate partial and final, Sort, WholeStageCodegen *(n), AQEShuffleRead coalesced/skewed/local, ShuffleQueryStage, InMemoryTableScan, Python UDFs…), where the stage boundaries are, what was pushed down to the reader, and red flags (cartesian products, single-partition exchanges, Python UDFs). Nothing leaves the page.

Partitions & AQE — which partition a key goes to (pmod(hash(key), n) with Spark’s Murmur3, checkable against SELECT hash(…)), how AQE would coalesce or split a shuffle of given sizes, and how many scan tasks a table of given size and file count gets.

The front end. A PySpark-flavoured DataFrame language (spark.table, filter/where, select, withColumn, groupBy().agg(), join(other, on, how), repartition, coalesce, orderBy, limit, distinct, hint, broadcast(), cache/persist(StorageLevel.…) and the actions count, show, collect, take, write, explain(mode)) and a SQL subset (SELECT … FROM … JOIN … WHERE … GROUP BY … HAVING … ORDER BY … LIMIT, /*+ BROADCAST(t) */ hints, EXPLAIN [EXTENDED|FORMATTED|COST]). Errors read like Spark’s (UNRESOLVED_COLUMN.WITH_SUGGESTION, AMBIGUOUS_REFERENCE, MISSING_AGGREGATION…).

Analysis and Catalyst. Names resolve to attributes with expression ids (amount#4); implicit casts are added. The optimizer runs ConstantFolding, BooleanSimplification, PruneFilters, InferFiltersFromConstraints (the isnotnull(…) filters), PushDownPredicates (through projections, aggregates on grouping columns and into the sides of joins), CombineFilters, ColumnPruning, CollapseProject and RemoveNoopOperators to a fixed point, recording each change (Catalyst, Armbrust et al., Spark SQL, SIGMOD 2015).

Size estimates follow SizeInBytesOnlyStatsPlanVisitor (no cost-based optimizer): a projection scales the size by row width, a filter does not reduce it, joins multiply, using Spark’s default type sizes.

Physical planning. FileSourceStrategy (scans with PushedFilters, DataFilters and the pruned ReadSchema), two-phase hash aggregation (and the four aggregates of count(distinct)), JoinSelection (hints; broadcast hash join under spark.sql.autoBroadcastJoinThreshold; shuffled hash join when preferSortMergeJoin=false and a side is small enough and 3× smaller; otherwise sort-merge; nested loop or cartesian without equality), EnsureRequirements (an Exchange only where the required distribution is not already met, sorts under sort-merge joins), CollapseCodegenStages for *(n), TakeOrderedAndProject for orderBy().limit(), CollectLimit for show. See Performance tuning.

Execution. Stages are cut at shuffles and broadcasts and numbered as the DAGScheduler does (parents first); scan partitions come from FilePartition.maxSplitBytes (spark.sql.files.maxPartitionBytes, openCostInBytes, default parallelism); shuffle partition sizes from pmod(Murmur3Hash(keys, 42), n) over the key distributions; map outputs live with the executor that wrote them and are tracked by the driver; reducers fetch them (locally or over the network).

Memory (memory management overview): 300 MiB reserved, spark.memory.fraction (0.6) of the rest unified, of which storageFraction (0.5) is storage that execution cannot evict; each of N running tasks gets between 1/2N and 1/N of the execution pool; what does not fit spills (sorts, aggregations) or fails (the hash table of a shuffled hash join, a broadcast relation). Cached blocks are stored per storage level, evicted least-recently-used (never for blocks of the same dataset), dropped (MEMORY_ONLY) or written to disk (MEMORY_AND_DISK) and recomputed from lineage when missing. memoryOverhead = max(384 MiB, 0.10 × heap) bounds memory outside the heap: exceed it and the container is killed.

Adaptive query execution (AQE): after each shuffle stage, CoalesceShufflePartitions (target = max(minPartitionSize, min(advisoryPartitionSizeInBytes, total / default parallelism))), OptimizeSkewedJoin (skewed if larger than both 5 × the median and 256 MiB; split along map outputs to about max(advisory, mean of the others)), and switching a sort-merge join to a broadcast hash join with local shuffle reads when a side turns out small.

Failures (Zaharia et al., Resilient Distributed Datasets, NSDI 2012): a lost executor’s running tasks are retried, its shuffle files and cached blocks are gone; reducers get FetchFailed, the stage is marked failed and only the missing map partitions are recomputed from lineage; spark.task.maxFailures (4) aborts a job; container kills for memory count as task failures. Speculation (spark.speculation.quantile 0.75, multiplier 1.5). Dynamic allocation (docs): requests after schedulerBacklogTimeout, doubling each round, release after executorIdleTimeout, keeping executors with cached data.

Hashing. Spark’s Murmur3_x86_32 with seed 42 (hashInt, hashLong, and the legacy hashUnsafeBytes for strings), checked against the documented values hash(1) = -559580957 and hash('Spark', array(123), 2) = -1321691492.

Data is modelled from statistics, not stored: row counts and sizes are computed, the rows shown are from a small sample (joins and filters on the sample can differ from the full data). No real files, no Python, no UDFs (a scenario fakes a Python worker’s memory with native memory per task).

A subset of Spark SQL: no window functions, subqueries, unions, set operations, outer-join reordering, bucketing, dynamic partition pruning, cost-based optimizer or table statistics (ANALYZE TABLE). Count distinct supports a single distinct aggregate. Plans follow Spark 3.5/4.0 formats; expression ids and plan ids are numbered from this page, not your session.

Task costs come from fixed per-core rates (scan 120 MB/s, network 110 MB/s, disk 220 MB/s, nanoseconds per row per operator), so durations are illustrative. No GC, no locality waits (spark.locality.wait), no barrier or fair scheduling, no external shuffle service or decommissioning; the shuffle files of a finished query are treated as cleaned up for dynamic allocation. Some timeouts are shorter than Spark’s defaults (each setting shows Spark’s).

AQE switches a join to broadcast once both map stages have finished (Spark may decide as soon as the small side finishes), and reads coalesced and skew-split partitions as described in ShufflePartitionsUtil, with the small-partition merge simplified.