Spark from the inside

You can write Spark code for months by copying snippets. Then one day a job that took 4 minutes takes 4 hours, or your Kafka stream is 3 million records behind, and the snippets stop helping. What helps is knowing the machine: who runs your lambda, when it runs, and what moves across the network.

So every chapter explains one mechanism and gives you a diagram you can poke. Each diagram has a 💥 button that breaks something on purpose — because you learn a system fastest by watching it fail.

The guide has levels. Chapters and blocks carry a chip — L1 essentials, L2 practitioner, L3 internals. Choose your depth with the switch in the bottom-left corner; deeper material hides until you want it. The map below jumps anywhere and ticks off what you've read.

Facts checked against the Apache Spark 4.2.0 documentation and the Spark source code (tag v4.0.0) in September 2026. Code examples are Java. DStreams (Spark Streaming) is the legacy streaming engine: the docs say it gets no more updates and recommend Structured Streaming for new pipelines. You still meet DStream jobs in production — this guide is for running and understanding them.

A Spark application is two kinds of processes. The driver runs your main() and owns the SparkContext. It plans the work and hands it out. The executors are JVM processes on worker nodes. They run tasks in threads and keep data in memory or on disk for your application.

Between them sits a cluster manager — Spark's own standalone manager, YARN or Kubernetes. The driver asks it for executors; after that, the driver talks to the executors directly. Each application gets its own executors, which live as long as the application does. Two applications therefore cannot share cached data without writing it to external storage.

The deploy mode decides where the driver runs: in cluster mode the framework launches it inside the cluster; in client mode it runs where you ran spark-submit. Either way, executors must be able to connect back to the driver for its whole life, so keep the driver close to the workers. Each driver serves a web UI, typically on port 4040.

An RDD (Resilient Distributed Dataset) is an immutable collection split into partitions. One task processes one partition. You never change an RDD; each transformation (map, filter, reduceByKey…) returns a new RDD that remembers its parent and the function to apply.

Transformations are lazy: Spark only records them. Work starts when an action — count(), collect(), saveAsTextFile() — needs a result. The chain of recorded steps is the lineage, and it is the "resilient" in RDD: if an executor dies and a partition is lost, Spark recomputes just that partition from its lineage.

SparkConf conf = new SparkConf().setAppName("word-count");
JavaSparkContext sc = new JavaSparkContext(conf);

JavaRDD<String> lines = sc.textFile("hdfs:///logs/*.txt");       // nothing is read yet
JavaPairRDD<String, Integer> counts = lines
    .flatMap(l -> Arrays.asList(l.split(" ")).iterator())      // narrow
    .mapToPair(w -> new Tuple2<>(w, 1))                           // narrow
    .reduceByKey(Integer::sum);                                   // wide: needs a shuffle

counts.cache();                        // a marker, still nothing computed
long distinct = counts.count();        // action → job 0, fills the cache
List<Tuple2<String, Integer>> some = counts.take(10);  // job 1, reads cached partitions
System.out.println(counts.toDebugString());          // prints the lineage

Each action submits a job. The job is split into stages, and each stage into tasks — one task per partition. Where does one stage end? At a shuffle.

Dependencies come in two kinds. Narrow: each output partition needs one parent partition (map, filter, flatMap). Spark pipelines narrow steps inside one task — a record flows through flatMap → mapToPair without any intermediate collection. Wide (shuffle): an output partition needs data from many parent partitions (reduceByKey, groupByKey, join, repartition). That needs an all-to-all exchange, so the stage must finish writing before the next can read: a barrier.

A shuffle moves every record to the partition that owns its key. The Spark docs call it expensive because it costs disk I/O, serialization and network I/O at once. It runs as two sets of tasks, named after MapReduce: map tasks organize their output by target partition, reduce tasks fetch and aggregate it.

On the map side, results are held in memory until they no longer fit, then sorted by target partition and written to a single file per map task (plus an index telling where each target partition's block starts). Each reduce task reads its block from every map task's file. These shuffle files stay on disk until the RDDs that need them are garbage collected — so a lost executor does not always mean a full recompute, and a long-running job can fill spark.local.dir.

Executor memory has two main customers. Execution memory is for computation — shuffles, joins, sorts, aggregations. Storage memory is for cached data and data propagated across the cluster, like broadcasts. Since the unified memory manager, both share one region M, and each can borrow the other's free space.

The rule that matters: execution may evict storage, but only until storage falls back to a protected region R inside M. Storage never evicts execution. If execution still needs more, it spills to disk.

With RDDs, Spark runs your lambdas as written — it cannot look inside them. A DataFrame query is different: it is a tree of relational operators Spark understands. So the Catalyst optimizer can rewrite it before anything runs. df.explain(true) prints the four phases: Parsed Logical Plan → Analyzed Logical Plan (names and types resolved against the catalog) → Optimized Logical Plan (rule-based rewrites) → Physical Plan (concrete operators such as a sort-merge join).

Two more layers work below the plan. Whole-stage code generation compiles a subtree of supported operators into a single Java function, so rows are not passed through one virtual call per operator. And Adaptive Query Execution (AQE), on by default (spark.sql.adaptive.enabled = true), re-optimizes the plan during execution using real statistics from finished shuffle stages.

Spark Streaming does not process records one by one. It cuts time into batch intervals (say 10 s). Everything that arrived during an interval becomes one RDD. A DStream (discretized stream) is just that series of RDDs, and every operation on a DStream is applied to each RDD in it.

So everything from Part I still holds — every batch is a normal Spark job with stages, tasks and shuffles. And like RDDs, DStreams are lazy: only output operations (foreachRDD, print, saveAs…Files) make Spark do the work. No output operation, no processing. By default, output operations run one at a time, in the order you defined them.

The Streaming tab of the web UI shows the two numbers that matter. Processing time is how long each batch took. Scheduling delay is how long a batch waited in the queue for earlier batches to finish. If processing time is consistently above the batch interval, or scheduling delay keeps growing, the application is falling behind.

stream.foreachRDD(rdd -> {                   // driver
  rdd.foreachPartition(records -> {           // executor, once per partition
    Connection c = ConnectionPool.get();      // your own lazily created pool
    while (records.hasNext()) c.send(records.next());
    ConnectionPool.release(c);
  });
});

The Kafka integration for Spark Streaming is spark-streaming-kafka-0-10 (for Kafka brokers 0.10.0+). It is a direct stream: there is no long-running receiver. At every batch the driver looks up the latest offsets for each partition and defines the batch as a set of offset ranges — "partition 0 from 1 200 to 1 450". The resulting KafkaRDD has exactly one Spark partition per Kafka partition, and executors read their range from Kafka when the task runs.

SparkConf conf = new SparkConf().setAppName("orders-stream");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));

Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "broker1:9092,broker2:9092");
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "orders-stream");         // one group.id per stream
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", false);          // we commit ourselves (next chapter)

JavaInputDStream<ConsumerRecord<String, String>> stream =
    KafkaUtils.createDirectStream(
        jssc,
        LocationStrategies.PreferConsistent(),
        ConsumerStrategies.<String, String>Subscribe(List.of("orders"), kafkaParams));

stream.mapToPair(r -> new Tuple2<>(r.key(), r.value()))
      .foreachRDD(rdd -> { /* output */ });

jssc.start();
jssc.awaitTermination();

Depend only on org.apache.spark:spark-streaming-kafka-0-10_2.13; the docs warn not to add kafka-clients yourself, because mismatched versions fail in hard-to-diagnose ways. Mark spark-core and spark-streaming as provided.

Spark output operations are at-least-once: after a failure, a batch may be written again. What you get end to end depends on where and when you store the offsets. The docs list three options, in order of increasing reliability and code complexity — plus one trap.

// Option 2: commit to Kafka after the output is stored
stream.foreachRDD(rdd -> {
  OffsetRange[] ranges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();  // first call on the stream!
  rdd.foreachPartition(records -> {
    OffsetRange o = ranges[TaskContext.get().partitionId()];            // this task's range
    writeIdempotently(records);                                        // e.g. upsert by key
  });
  ((CanCommitOffsets) stream.inputDStream()).commitAsync(ranges);
});

Map<TopicPartition, Long> from = db.loadOffsets("orders-stream");      // your table
JavaInputDStream<ConsumerRecord<String, String>> stream =
    KafkaUtils.createDirectStream(jssc, LocationStrategies.PreferConsistent(),
        ConsumerStrategies.<String, String>Assign(from.keySet(), kafkaParams, from));

stream.foreachRDD(rdd -> {
  OffsetRange[] ranges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
  List<Result> results = compute(rdd);          // e.g. an aggregate collected to the driver
  db.inTransaction(tx -> {
    tx.save(results);
    // update only where stored offset == range.fromOffset(); otherwise roll back
    tx.advanceOffsets(ranges);
  });
});

Window operations compute over the last N batches. Two parameters: window length (how much time) and sliding interval (how often). Both must be multiples of the batch interval — the source code rejects the DStream otherwise. reduceByKeyAndWindow(f, Durations.seconds(30), Durations.seconds(10)) gives per-key totals over 30 s, every 10 s.

With an inverse function — reduceByKeyAndWindow(f, invF, 30 s, 10 s) — Spark updates the previous window result: add what entered, subtract what left. Cheaper for long windows, but it needs checkpointing. Windowed DStreams are persisted in memory automatically.

State across the whole stream: updateStateByKey(func) calls your function every batch for every key, with the new values and the previous state. It requires a checkpoint directory.

Two kinds of checkpoint

String checkpointDir = "hdfs:///checkpoints/orders-stream";

JavaStreamingContext ctx = JavaStreamingContext.getOrCreate(checkpointDir, () -> {
  JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
  // create ALL DStreams and output operations here
  jssc.checkpoint(checkpointDir);
  return jssc;
});
ctx.start();
ctx.awaitTermination();

If the directory exists, the context is rebuilt from it; otherwise the function creates a fresh one. The driver must also be restarted automatically after a failure — that is the job of your deployment (cluster mode supervision, Kubernetes, YARN attempts), not of getOrCreate.

A direct stream, left alone, takes everything between the last offset and the latest one in a single batch. After a restart with an hour of backlog, that one batch is enormous. Two settings bound it. spark.streaming.kafka.maxRatePerPartition is a hard ceiling in records per second per partition, so the batch size is at most rate × partitions × batch seconds. spark.streaming.backpressure.enabled makes Spark estimate, from recent batch delays and processing times, how fast it can go, and set the rate dynamically — still capped by maxRatePerPartition if set.

Cheat sheet