The RDD Escape Hatch

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.

orders = spark.read.parquet("/data/lake/orders")

totals = (
    orders.rdd
    .filter(lambda r: r.status == "Completed")
    .map(lambda r: (r.region, r.profit))
    .reduceByKey(lambda a, b: a + b)
    .toDF(["region", "total_profit"])
)

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.

The cost of the .rdd wall

• Every column is read: The lambdas never read the fourth column, but Spark cannot see that, so it produces all 4 columns of every row. On a 120-column table, all 120 come along.
• Every row is converted: Each row leaves Spark's compact internal format and becomes a Python Row object before a lambda can run on it.
• The filter stays put: The status test runs inside Python once that conversion is done. Catalyst can neither move it earlier nor combine it with another filter.
• The plan goes dark: After toDF, the new DataFrame starts from an existing RDD. Its plan shows none of the steps that built it, including the shuffle inside reduceByKey.

A job reads a 120-column table, calls .rdd, and its lambdas read 2 fields. How many columns does Spark produce for each row it hands to the lambdas?

• 2, because Spark traces which fields the lambdas access
• 3, the 2 fields used plus a row index Spark attaches
• 120 for the first row, then 2 once it sees the fields used
• 120, because Catalyst cannot see which fields the lambdas read

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:

completed = (
    orders
    .filter(F.col("status") == "Completed")
    .select("region", "profit")
)

totals = (
    completed.rdd
    .map(lambda r: (r.region, r.profit))
    .reduceByKey(lambda a, b: a + b)
    .toDF(["region", "total_profit"])
)

Now only Completed rows cross the wall, carrying 2 columns each, so far fewer Python objects get built. The better fix keeps the job on the near side of the wall entirely, since each of these lambdas has a DataFrame equivalent.

Translating lambdas back to columns

Jobs that need the RDD API

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

• Checking layout: df.rdd.getNumPartitions() returns a fresh DataFrame's partition count without reading rows.
• Setup per partition: mapPartitions opens 1 connection or loads 1 model per partition, not once per row.
• Libraries that only accept whole Python objects, called on rows a DataFrame step has already filtered and trimmed.
• Legacy code mid-migration, kept as an RDD step until each lambda is rewritten as a column expression.

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.

About this lesson

This section is part of the Catalyst Optimizer: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.