DataFrames and Spark SQL

"I used to micromanage every map and filter by hand. Then the optimizer read my query, rearranged my whole life, and somehow I ran twice as fast doing half the work."

An RDD That Learned to Let Go

The Resilient Distributed Dataset of Section 7.2 is a powerful and honest abstraction: a partitioned collection of arbitrary Python or Scala objects, transformed by functions you supply. Its honesty is also its limitation. Because the elements are opaque objects and the transformations are opaque closures, the engine cannot see inside them. It cannot know that your map only ever reads two of the twenty fields in each record, that your filter would eliminate ninety percent of the rows if it ran first, or that two of your joins commute. It must run your steps in the order you wrote them, materializing whatever you told it to materialize. The RDD does what you say, which means the quality of the plan is entirely on you.

The DataFrame changes the abstraction in one decisive way: the data now has a schema. A DataFrame is a distributed table of rows with named, typed columns, exactly the shape of the tabular features that classical models in Chapter 12 consume. Once Spark knows the columns and their types, and once your computation is expressed as relational operations over those columns (select, filter, join, group) rather than as opaque closures, the engine can read, analyze, and rewrite your query before running it. That rewrite engine is Catalyst, and the structure is what makes it possible.

1. From Imperative RDDs to Declarative DataFrames Beginner

The clearest way to feel the difference is to write the same computation twice. Suppose we have event records and we want, per country, the total spend of users on mobile devices. With RDDs you script the dataflow: parse each line, filter to mobile, project the fields you need, key by country, and reduce. Every decision about order and intermediate shape is yours, and Spark executes it verbatim.

# RDD style: you specify every step and its order.
rdd = sc.textFile("events.csv")
result = (rdd
    .map(lambda line: line.split(","))                       # parse every column
    .filter(lambda r: r[3] == "mobile")                      # keep mobile rows
    .map(lambda r: (r[1], float(r[5])))                       # (country, spend)
    .reduceByKey(lambda a, b: a + b))                         # sum per country

Now the DataFrame form. You declare the table, the predicate, the grouping, and the aggregate, and you say nothing about order or intermediate representation. The equivalent Spark SQL string expresses the identical intent in the language a data analyst already knows.

# DataFrame style: you declare the result; Spark plans the execution.
df = spark.read.csv("events.csv", header=True, inferSchema=True)
result = (df
    .filter(df.device == "mobile")        # a relational predicate, not a closure
    .groupBy("country")
    .sum("spend"))

# The same query as Spark SQL over the same engine and the same optimizer.
df.createOrReplaceTempView("events")
result = spark.sql("""
    SELECT country, SUM(spend) AS total_spend
    FROM events
    WHERE device = 'mobile'
    GROUP BY country
""")

The two listings compute the same answer, but they are not the same instruction to the engine. Code 7.3.1 is a recipe Spark must follow literally. Code 7.3.2 is a description of the goal that Spark is free to achieve any correct way it likes, and "any correct way it likes" is precisely the freedom an optimizer needs. The DataFrame and the SQL string in Code 7.3.2 are interchangeable surfaces over one engine: both lower to the same logical plan, so everything we say about the optimizer applies equally to the analyst writing SQL and the engineer writing Python.

2. What Catalyst Actually Does Intermediate

Catalyst is a tree-rewriting system. It represents your query as a logical plan, a tree of relational operators (scan, filter, project, join, aggregate), and applies rules that transform the tree into an equivalent tree with lower cost. Three rewrites carry most of the benefit, and all three exploit the schema that RDDs lack.

Predicate pushdown moves filters as close to the data source as possible. If your query reads a table and then keeps only the rows where country = 'US', Catalyst pushes that predicate into the scan itself, so a Parquet or ORC reader skips entire row groups whose statistics prove no row can match, and the network and CPU never touch rows that will be discarded. Column pruning drops columns that no downstream operator references; if your final projection needs two of twenty columns, the scan reads only those two (plus any the pushed filter requires), which on columnar storage means reading a small fraction of the file. Join reordering rearranges a chain of joins so that the most selective, smallest intermediate results are produced first, shrinking the data that flows into later joins. Spark also folds constants, collapses adjacent projections, and, when statistics are available, makes cost-based choices such as switching a shuffle join to a broadcast join when one side is small.

The crucial point is that pushdown and pruning change how much data is read and moved, and in a distributed system data movement is the dominant cost, the same shuffle cost we traced through MapReduce in Chapter 6 and will quantify against the alpha-beta model of Chapter 3. To make the rewrites concrete rather than abstract, the code below implements a miniature optimizer over a tiny logical plan. It applies predicate pushdown and column pruning to a declared query and reports the rows and cells scanned before and after, the same accounting Catalyst does at scale.

"""Tiny logical-plan optimizer: predicate pushdown + column pruning."""

TABLE = "events"
COLUMNS = ["user_id", "country", "ts", "device", "clicks", "spend"]
ROWS_ON_DISK = 1_000_000
FILTER_SELECTIVITY = 0.18                       # fraction surviving country == 'US'

# The DECLARED query, bottom-up: Scan -> Filter -> Project.
declared_plan = [
    ("Scan",    {"table": TABLE, "columns": COLUMNS}),
    ("Filter",  {"predicate": "country == 'US'", "selectivity": FILTER_SELECTIVITY}),
    ("Project", {"columns": ["user_id", "spend"]}),
]

def cost_of(plan):                              # rows and cells touched at the scan
    scan = plan[0][1]
    rows = scan.get("scan_rows", ROWS_ON_DISK)
    cols = len(scan["columns"])
    return rows, cols, rows * cols

def predicate_pushdown(plan):                   # let the scan read only matching rows
    out = [list(op) for op in plan]
    flt = next((op[1] for op in out if op[0] == "Filter"), None)
    if flt is not None:
        out[0][1]["scan_rows"] = int(ROWS_ON_DISK * flt["selectivity"])
        out[0][1]["pushed_predicate"] = flt["predicate"]
    return out

def column_pruning(plan):                       # keep only columns someone reads
    out = [list(op) for op in plan]
    needed = set()
    for op in out:
        if op[0] == "Project":
            needed.update(op[1]["columns"])
        if op[0] == "Filter":
            needed.add(op[1]["predicate"].split(" ")[0])   # column in the predicate
    out[0][1]["columns"] = [c for c in out[0][1]["columns"] if c in needed]
    return out

def report(label, plan):
    rows, cols, cells = cost_of(plan)
    print(f"{label:<26} rows={rows:>9,}  cols={cols}  cells={cells:>11,}")
    return cells

print("Query: SELECT user_id, spend FROM events WHERE country = 'US'\n")
base = report("unoptimized plan", declared_plan)
opt  = column_pruning(predicate_pushdown(declared_plan))
done = report("optimized plan", opt)
print(f"\npushed predicate : {opt[0][1].get('pushed_predicate')}")
print(f"pruned columns   : {opt[0][1]['columns']}")
print(f"cells scanned    : {base:,} -> {done:,}")
print(f"data movement cut : {100 * (1 - done / base):.1f}%")

Query: SELECT user_id, spend FROM events WHERE country = 'US'

unoptimized plan           rows=1,000,000  cols=6  cells=  6,000,000
optimized plan             rows=  180,000  cols=3  cells=    540,000

pushed predicate : country == 'US'
pruned columns   : ['user_id', 'country', 'spend']
cells scanned    : 6,000,000 -> 540,000
data movement cut : 91.0%

Nothing about the answer changed; the query still returns the spend of every US user. What changed is the volume of data the engine reads and ships, and that volume fell by an order of magnitude from two rules a beginner would not think to apply by hand. This is the mechanism behind the headline claim of the section: because Catalyst applies these rewrites uniformly and the RDD programmer usually does not, the structured path tends to win.

3. Why DataFrames Usually Beat Hand-Written RDD Code Intermediate

Optimization is only half the story. Once Catalyst has chosen a physical plan, a second engine, Tungsten, executes it efficiently. Tungsten stores rows in a compact, cache-friendly binary layout rather than as boxed JVM objects, manages memory off the garbage-collected heap, and generates Java bytecode at runtime that fuses a whole chain of operators into a single tight loop (whole-stage code generation). The result is that a DataFrame pipeline often runs as a handful of compiled loops over packed memory, while the equivalent RDD pipeline runs as a chain of virtual function calls over scattered objects. Structure pays twice: once when Catalyst shrinks the work, and again when Tungsten executes what remains close to the metal.

There is a productivity argument layered on top of the performance one. The hand-written RDD in Code 7.3.1 hard-codes a column order and a parsing scheme; change the input and it breaks silently. The DataFrame in Code 7.3.2 carries a schema, so a type mismatch is caught at analysis time, before the job launches across the cluster. For the tabular feature engineering that precedes nearly every model, joining user tables to event tables, aggregating to per-entity features, encoding categoricals, this combination of safety and speed is why DataFrames, not RDDs, are the default surface for data preparation in production AI pipelines.

df = spark.read.parquet("events.parquet")          # columnar source with statistics
q  = (df.filter(df.country == "US")
        .select("user_id", "spend"))               # declare; do not micromanage

q.explain(mode="formatted")
# In the printed physical plan you will see, with no extra work on your part:
#   PushedFilters: [EqualTo(country, US)]          <- predicate pushdown
#   ReadSchema: struct<user_id,spend>              <- column pruning

4. Structured Tables Meet Columnar Storage Advanced

Predicate pushdown and column pruning deliver their full benefit only when the storage layer can act on them, and that is the bridge from this section to the next chapter. A row-oriented file (one row's fields stored contiguously) cannot read column spend without also reading every other field of that row, so column pruning saves little. A columnar file such as Parquet or ORC stores each column contiguously and ships per-column statistics (min, max, null counts) for each block, so pruning reads only the requested columns and pushdown skips whole blocks whose statistics rule out a match. The 91 percent reduction in Output 7.3.3 is a paper exercise on a row store; on columnar storage it is roughly what you actually observe, because the format was designed to let the optimizer's decisions translate into bytes never read from disk.

This is why structured Spark and columnar storage are a matched pair, and why Chapter 8 takes up distributed storage and data loading directly after this chapter. The DataFrame schema, the Catalyst plan, and the Parquet layout are three views of the same idea: tell the system the structure of your data, and every layer from the query planner to the disk reader can exploit it. For AI workloads this matters twice over, because the feature tables you build here become the input pipeline that feeds the training loop, and a reader that skips 90 percent of the bytes is a training job that starts 90 percent sooner.

We have now seen the structured API in full: the schema that makes optimization legal, the Catalyst rewrites that shrink the work, the Tungsten engine that runs the remainder close to the metal, and the columnar storage that turns the optimizer's decisions into bytes never read. One question we have deferred is when all this planning happens. Catalyst rewrote a plan in Code 7.3.3, but a real Spark job builds that plan lazily, accumulating transformations without running them until an action forces execution, which is exactly what gives the optimizer a complete query to rewrite. That lazy, plan-then-execute model and the directed acyclic graph it produces are the subject of Section 7.4.