Catalyst Optimizer: Beginner

Here is a single question, total profit of completed orders per region, in both styles. orders_rdd holds tuples of (order_id, status, region, profit), and orders is a DataFrame with those same 4 columns.

Run either chain and you get the same totals. The difference is what Spark knows before running them. To Spark, lambda row: row[1] == "Completed" is a compiled Python function. Spark can call it on an element and get a value back, and that is all it can do. It can't tell which field the function reads, that the test is an equality, or that position 1 holds a status string. The DataFrame filter hands Spark the expression itself: = applied to the column status and the literal Completed.

If Spark can read a step, it can change that step. The DataFrame chain is built from expressions over named columns, so Spark passes the whole plan through its optimizer, called Catalyst, before scheduling the first task. Catalyst notices that order_id is never used and leaves it unread, and it chooses how the grouping is carried out. The RDD chain gets none of this. Spark runs each lambda exactly where you placed it, on every field of every element, in the written order.

You already know Spark waits for an action before doing any work. That wait is what gives Catalyst room to act. When you call collect() or count(), Spark holds the full chain from the read to the result, so it can optimize all of it at once. With DataFrames, a messy chain of 10 steps and a tidy chain of 4 can come out as the same plan. With RDDs, 10 steps run as 10 steps.

Run the DataFrame version on the practice orders table. The extra filter drops orders that have no region, which would otherwise form a group of their own. Then change "Completed" to "completed". The query still runs, but every group disappears, because string comparison is case-sensitive.

from pyspark.sql import functions as F

orders = spark.table("orders")

df_result = (
    orders
    .filter(F.col("status") == "Completed")
    .filter(F.col("region").isNotNull())
    .groupBy("region")
    .agg(
        F.count("*").alias("completed_orders"),
        F.round(F.sum("profit"), 2).alias("total_profit"),
    )
    .orderBy("region")
)
df_result.show()

Read literally, this chain sorts every order and then throws most of them away. A global sort needs a shuffle: every row crosses the network so that each task owns a slice of the sorted range. Catalyst sees that the filter reads only status. That column exists before withColumn and the sort leaves it untouched, so testing it first can't change which rows come out. Catalyst moves the filter below the sort and below withColumn, right after the read. Say orders holds 8,000,000 rows and 2,000,000 of them are Completed. As written, the sort would shuffle all 8,000,000. After the rewrite it shuffles 2,000,000.

A filter moves below a step only when that step can't affect its answer. Put a filter on region after groupBy("region") and it moves below the grouping, because removing a region's rows before grouping removes exactly that region's group. A filter on total_profit > 5000 after the same grouping stays where it is. total_profit does not exist until the grouping has run, and a rule cannot test a value nobody has computed yet.

You can apply the rule yourself. A filter runs earlier only if it can be written in columns that exist at that earlier point. A derived column qualifies when its formula reads such columns, so Catalyst rewrites profit_k > 2 as profit / 1000 > 2 and moves it too. An aggregate's result never qualifies, because computing it takes every row of the group.

In the next animation the query averages price by category over a 6-column products table. Catalyst dropped product_id and product_name before the first task ran, along with supplier and weight_kg, so none of those 4 columns crosses the shuffle. Kitchen comes out at (45 + 60 + 35) / 3, about 46.67.

Every rewrite returns the rows your written chain would have returned. That's why Catalyst applies them without asking you. Its limit is that it can only rewrite what it can read. It cannot remove a shuffle your logic requires: a groupBy still groups and an orderBy still sorts. It also can't see inside a Python function. Write the query in whatever order reads clearest, and spend your effort on the steps Catalyst cannot change.

Run this version, where every filter comes last. Then move the status and region filters up to sit right after spark.table("orders") and run again: the rows match. Now try moving the profit_k filter up as well. Spark rejects it at once, because profit_k does not exist at that point in your code. Catalyst can make that move by rewriting the filter in terms of profit, but your code cannot.

from pyspark.sql import functions as F

orders = spark.table("orders")

late_filters = (
    orders
    .withColumn("profit_k", F.round(F.col("profit") / 1000, 3))
    .select("order_id", "status", "region", "profit_k")
    .orderBy("order_id")
    .filter(F.col("status") == "Completed")
    .filter(F.col("region").isNotNull())
    .filter(F.col("profit_k").isNotNull())
)
late_filters.show()

Every DataFrame has an .rdd attribute. It returns the same data as an RDD of Row objects, and from there map accepts any Python function, as do filter and reduceByKey. Engineers reach for it when a step reads more naturally as a lambda, or out of habit carried over from older code.

This produces the totals you saw in the DataFrames over RDDs section, at a higher price. The expression orders.rdd is a wall in the plan. Catalyst optimizes what lies before the wall, here the bare read, and then hands every row across it. Everything past the wall stays out of Catalyst's view until toDF builds a fresh DataFrame from the result.

Where you place the wall decides how much you lose. DataFrame operations written before .rdd are still optimized. Filter and select first, and less crosses the wall:

Some jobs have no column-expression form, and those are what .rdd exists for:

This cell does the job of an old RDD program that counted products and found the top price per category with map and reduceByKey. Run it, then compare: F.max skips products whose price is NULL, while Python's max given a None would raise a TypeError and fail the task.

from pyspark.sql import functions as F

products = spark.table("products")

per_category = (
    products
    .filter(F.col("category").isNotNull())
    .groupBy("category")
    .agg(
        F.count("*").alias("n_products"),
        F.max("price").alias("max_price"),
    )
    .orderBy("category")
)
per_category.show()

Dataset is the typed API in Spark's compiled JVM languages. You declare a class, say Order with 4 fields, and get a distributed collection of Order objects. The compiler then rejects code that reads a field the class lacks. A DataFrame is the same structure with the generic Row type standing in for your class, which is why Spark's documentation describes a DataFrame as a Dataset of Row. Python has no compile step that could check field names, so it gets DataFrames only. Every DataFrame you have written in these lessons is this untyped form: rows whose column names Spark checks when each line runs, with no compiler involved.

Suppose a teammate types statuss for status. Here is where each of the 2 Python APIs stops them.

The RDD line is accepted without complaint, because Spark does not inspect a lambda when you define it. The mistake surfaces at count(), when a task first calls the lambda on a row and Python raises an AttributeError. If that filter sits after a 40-minute stretch of work in the same job, the job runs for 40 minutes before failing. The DataFrame line fails the moment it executes, with an AnalysisException that names statuss and points to the columns that do exist, and no job ever starts.

This holds for a standard Spark session, where your program runs inside the driver. A session that talks to a remote Spark server through a thin client can defer the check until the first action or the first schema lookup.

Everything so far happens inside Spark, out of sight. explain() prints the plan Catalyst produced for a DataFrame. It starts no job and reads no rows, so you can call it on a query over a huge table before you commit any cluster time.

The filter is written last, after the grouping. The printed plan is a tree, drawn with the result at the top and the read at the bottom. Data flows upward, so start at the bottom line. The numbers after # are internal ids that change from run to run, and long lines are trimmed here to their first part.

explain(mode="extended") prints 4 plans. The parsed plan comes first and is your chain as written. After the analyzed plan comes the optimized plan, which is what Catalyst turned it into, and the physical plan prints last. Comparing the parsed and optimized plans shows each rewrite in action.

On the left, the filter sits above the aggregate. On the right it has dropped below it and gained a NULL check, and a Project now keeps only region and profit. The relation still lists 4 columns at this level, and the physical planner turns that Project into the 2-column scan from the table above.

Call explain() on the totals DataFrame from the escape-hatch section, and you get a short plan whose bottom line reads Scan ExistingRDD[region#20,total_profit#21]. The 3 lambdas and the shuffle inside reduceByKey are all missing, because Catalyst received the RDD as a finished source. A plan that starts at Scan ExistingRDD tells you 2 things: the steps between .rdd and toDF were not optimized and don't appear here, and counting its Exchange lines undercounts the job's shuffles. Any DataFrame work before .rdd was optimized on its own and isn't shown in this plan either.