Spark Architecture DataFrames and Lazy Evaluation

Published by Programming Line in Data Engineering in Depth. Last updated .

Purpose and Outcome

Spark lets a data engineer describe a distributed computation without manually assigning records to machines. In this lesson, the named mechanism is the DataFrame API backed by lazy evaluation. The outcome is practical: you should be able to look at a chain of DataFrame calls, predict when work actually starts, read the shape of the physical plan, and choose transformations that reduce unnecessary scans, shuffles, and recomputation.

A Spark application has a driver process and many executor processes. The driver runs your Python, Scala, SQL, or Java program. Executors store partitions of data, run tasks, and return status or results. A DataFrame is not the data itself. It is a distributed, schema-aware logical description of rows and columns. When you call filter, select, join, or groupBy, Spark records a new step in a plan. It does not immediately read all input files or move all rows. Actual execution begins only when an action asks for a result, such as show, count, collect, or write.

How Spark Turns DataFrames Into Work

The internal path is a pipeline. First, your code builds an unresolved logical plan. Column names, function calls, and table references may still be symbolic. The analyzer resolves those symbols using schemas, catalogs, function registries, and type rules. The result is an analyzed logical plan. Catalyst, Spark’s query optimizer, then rewrites that plan into an optimized logical plan. Common rewrites include predicate pushdown, column pruning, constant folding, null handling simplification, and join reordering when statistics are available.

After optimization, Spark selects a physical plan. A physical plan names concrete operators such as file scans, projections, filters, hash aggregates, sorts, exchanges, and joins. An Exchange is especially important because it means Spark must repartition data across the cluster. Exchanges create shuffle boundaries. The scheduler then splits the physical plan into stages. Within each stage, tasks can run independently over partitions. At shuffle boundaries, downstream tasks wait for upstream shuffle files to be produced.

Lazy evaluation matters because Spark can optimize across the whole chain of transformations. If you select two columns from a wide Parquet table, then filter rows, then aggregate, Spark can often read only the required columns and push parts of the filter into the file scan. If each line executed immediately, those cross-step optimizations would be much weaker.

API Anatomy

DataFrame code combines transformations and actions. Transformations return another DataFrame and stay lazy. Narrow transformations, such as select, withColumn, and many filter operations, can process each input partition without requiring rows from other partitions. Wide transformations, such as groupBy, many join operations, distinct, and global orderBy, usually require a shuffle because rows with the same key must meet in the same partition.

Actions materialize a result. show prints a sample to the driver. count returns a scalar. collect brings all rows to the driver and is risky for large data. write materializes output files. explain is a diagnostic action-like method: it prints the current plan and helps you see what Spark intends to do before you commit to an expensive run.

Example 1: Projection and Filter Stay Lazy

The following example creates a small DataFrame, adds a derived column, filters to books, and selects two columns. Until show runs, Spark has only a plan. The formatted plan will include a scan of the existing rows, a projection for revenue, and a filter for sku = book. The deterministic result is two rows.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("lazy-filter-demo").getOrCreate()
orders = spark.createDataFrame(
    [
        (1, "book", 2, 12.50),
        (2, "pen", 10, 1.20),
        (3, "book", 1, 12.50),
        (4, "desk", 1, 199.00),
    ],
    ["order_id", "sku", "qty", "unit_price"],
)

priced = orders.withColumn("revenue", F.col("qty") * F.col("unit_price"))
books = priced.filter(F.col("sku") == "book").select("order_id", "revenue")

books.explain(mode="formatted")
books.show()
# Expected rows, order may vary:
# +--------+-------+
# |order_id|revenue|
# +--------+-------+
# |       1|   25.0|
# |       3|   12.5|
# +--------+-------+

The design point is that priced and books are not cached tables. They are references to plans. If you call books.show() twice, Spark may recompute the plan twice unless the underlying data source or a cache prevents it.

Example 2: Join and Aggregation Introduce a Shuffle

This example joins orders to customers, groups by segment, and sums revenue. The join aligns rows with the same customer_id. The aggregation aligns rows with the same segment. On larger data, one or both operations may require exchanges. The exact join strategy can vary based on statistics and settings, but the output values are fixed.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("shuffle-join-demo").getOrCreate()
orders = spark.createDataFrame(
    [(1, 10, 25.0), (2, 20, 12.0), (3, 10, 18.0), (4, 30, 7.0)],
    ["order_id", "customer_id", "revenue"],
)
customers = spark.createDataFrame(
    [(10, "retail"), (20, "business"), (30, "retail")],
    ["customer_id", "segment"],
)

by_segment = (
    orders.join(customers, "customer_id")
    .groupBy("segment")
    .agg(F.sum("revenue").alias("total_revenue"))
    .orderBy("segment")
)

by_segment.explain(mode="formatted")
by_segment.show()
# Expected rows:
# +--------+-------------+
# | segment|total_revenue|
# +--------+-------------+
# |business|         12.0|
# |  retail|         50.0|
# +--------+-------------+

If the customers table is small enough, Spark may broadcast it to executors and avoid shuffling both sides of the join. If neither side is small, Spark often repartitions both sides by the join key. The aggregation still needs rows for each segment to meet, so it usually performs partial aggregation before a shuffle and final aggregation after it.

Example 3: Cache a Reused Plan Deliberately

Lazy plans are recomputed for each action unless persisted. Caching is useful when the same expensive intermediate DataFrame feeds multiple actions. It is not free: cached partitions consume executor memory or disk, and a poor cache choice can evict more valuable data.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("cache-demo").getOrCreate()
events = spark.createDataFrame(
    [("u1", "click"), ("u1", "buy"), ("u2", "click"), ("u3", "view")],
    ["user_id", "event_type"],
)

active_users = events.groupBy("user_id").agg(F.count("*").alias("events"))
active_users.cache()

print(active_users.count())
active_users.filter(F.col("events") > 1).show()
active_users.unpersist()
# Expected count: 3
# Expected filtered row: u1 with 2 events

The first action, count, materializes the cache. The later filter can reuse cached aggregated user counts. unpersist releases the storage. In real pipelines, cache after expensive narrowing or aggregation, not immediately after reading massive raw input unless many downstream branches need exactly that input.

Design Choices and Trade-offs

Choose DataFrames when you want relational operations, schema awareness, optimizer support, and interoperability with SQL and file formats such as Parquet. Use lower-level RDDs only when your computation does not fit row and column operations. RDDs give more control over arbitrary objects, but they bypass many Catalyst optimizations.

Prefer built-in functions from pyspark.sql.functions over Python user-defined functions. Built-in expressions remain visible to Catalyst and can participate in code generation, predicate pushdown, and column pruning. Python UDFs often serialize data between the JVM and Python worker processes and hide expression details from the optimizer. Pandas UDFs can be efficient for vectorized operations, but they still introduce serialization and memory considerations.

Partitioning is another trade-off. Too few partitions underuse executors and make individual tasks large. Too many partitions increase scheduling overhead and create many small output files. Repartitioning can improve parallelism or align keys before a join, but it is a shuffle. Coalescing can reduce partitions with less movement, but it may reduce parallelism. The right choice depends on data size, cluster shape, downstream file layout, and key skew.

Failure Modes and Troubleshooting

Symptom: a job appears to do nothing while transformations run, then fails during write. Cause: transformations were lazy, so the invalid column, corrupt file, or type mismatch surfaced only at the action. Diagnose: call printSchema, inspect explain, and run limit(20).show() on the smallest DataFrame that should still contain the problem. Correct: fix column names, casts, or input parsing before the final action.

Symptom: one stage takes far longer than others and a few tasks run alone near the end. Cause: skewed keys, such as one customer or date owning a huge share of rows, made some shuffle partitions much larger. Diagnose: inspect the Spark UI stage page for task duration and shuffle read size variance, then group by the suspected key and count. Correct: filter impossible keys, salt hot keys, pre-aggregate before joins, or broadcast the smaller side when appropriate.

Symptom: the driver crashes or runs out of memory after a query. Cause: collect or conversion to Pandas pulled a distributed result into one process. Diagnose: search the job for collect, toPandas, and large show calls. Correct: write results to storage, use limit for inspection, or aggregate before returning small summaries.

Symptom: a cached DataFrame does not speed up later actions. Cause: the cache was never materialized, the plan differs from the cached plan, or storage eviction removed blocks. Diagnose: trigger a small action after cache, inspect the storage tab, and compare query plans. Correct: cache the exact reused DataFrame, materialize it once, and unpersist it after use.

Security, Performance, and Reliability Implications

Lazy DataFrames improve performance, but they also delay failures. Production jobs should validate schemas and input paths early with cheap actions when failure timing matters. Avoid printing sensitive data with show in shared logs. The Spark UI can expose SQL text, paths, and metrics, so access should match the sensitivity of the data platform.

Reliability comes from understanding lineage. Spark can recompute lost partitions from the lazy plan when an executor fails. Very long lineage chains, however, can make recovery expensive. For iterative workloads or multi-hour pipelines, checkpointing or writing stable intermediate tables can shorten lineage and create restart points. The trade-off is extra I/O and storage.

Hands-on Lab: Inspect and Materialize a Lazy Plan

Prerequisites: a local PySpark shell or notebook, Java configured for Spark, and permission to write under /tmp. The lab creates a tiny Parquet dataset, reads it, builds a lazy plan, explains it, writes output, and verifies the result.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("lesson-lab").getOrCreate()
spark.sparkContext.setLogLevel("WARN")

input_path = "/tmp/spark_lesson_events"
output_path = "/tmp/spark_lesson_daily_counts"

sample = spark.createDataFrame(
    [
        ("2026-09-01", "u1", "view"),
        ("2026-09-01", "u1", "click"),
        ("2026-09-01", "u2", "view"),
        ("2026-09-02", "u1", "buy"),
    ],
    ["event_date", "user_id", "event_type"],
)
sample.write.mode("overwrite").parquet(input_path)

events = spark.read.parquet(input_path)
daily = (
    events.filter(F.col("event_type").isin("view", "click", "buy"))
    .groupBy("event_date")
    .agg(
        F.count("*").alias("events"),
        F.countDistinct("user_id").alias("users"),
    )
    .orderBy("event_date")
)

daily.explain(mode="formatted")
daily.write.mode("overwrite").parquet(output_path)
spark.read.parquet(output_path).show()
# Expected rows:
# 2026-09-01: events=3, users=2
# 2026-09-02: events=1, users=1
  1. Run the script in a PySpark-enabled environment.
  2. Before the final write, read the formatted plan. Identify the scan, filter, aggregate, sort, and any exchange operators.
  3. Confirm that no output directory is created by the transformation chain alone. The directory appears when daily.write executes.
  4. Verify the displayed counts: September 1 has three events from two users, and September 2 has one event from one user.
  5. Cleanup by removing /tmp/spark_lesson_events and /tmp/spark_lesson_daily_counts from local or distributed storage.

If cleanup is performed from Python, use filesystem APIs appropriate to your environment. On a local machine, shell removal is enough. On HDFS or cloud object storage, use the platform command or SDK so you delete the actual stored paths, not just a local placeholder.

Assessment Exercises

  1. You have df.filter("country = 'US'").select("user_id") followed by count. Which parts are transformations, which part is the action, and what optimizations might Spark apply before reading a columnar file?
  2. A join between a 4 TB click table and a 5 MB campaign table is slow and shows large shuffle reads on both sides. What physical strategy would you try to encourage, and why?
  3. A notebook calls collect after grouping by user ID and crashes the driver. Redesign the inspection step so the computation stays distributed until the result is small.
  4. A cached DataFrame is used by three actions, but only the first action is slow and the next two are fast. Explain what happened in terms of lazy materialization.
  5. Given a plan with repeated Exchange operators, name two code or data layout changes that could reduce shuffle cost without changing the result.

Summary

Spark DataFrames are optimized descriptions of distributed row and column computations. Lazy evaluation lets Spark analyze a full chain of transformations, rewrite it with Catalyst, choose a physical plan, and execute stages only when an action requires a result. For data engineering, this model is central: it explains why small code changes can alter scans, shuffles, memory pressure, output files, and failure timing. Read plans, use built-in expressions, treat actions carefully, and persist intermediates only when reuse justifies the cost.