Lazy evaluation: transformations and actions

The PySpark examples in this wiki need a Spark cluster and do not run here. The site test checks that each one is valid Python (py_compile); the output, when it appears, is the one the documentation describes.

A transformation builds, an action executes

from pyspark.sql import functions as F

orders = spark.read.table("silver.orders")               # nothing read yet
delivered = orders.where(F.col("status") == "delivered") # still nothing
per_day = delivered.groupBy("order_date").agg(F.sum("amount").alias("total"))  # still nothing

per_day.count()   # action: now Spark optimizes the whole plan and runs it

Transformations (select, where, withColumn, join, groupBy().agg(), orderBy) return a new DataFrame and only add a step to the plan. Actions (count, show, collect, take, write, toPandas) ask for a result and trigger one or more *jobs*.

Why waiting is good

Seeing the whole plan before running, the optimizer (Catalyst) can:

• push the filter down to the read (*predicate pushdown*): Parquet/Delta skips whole blocks;
• read only the columns that are used (*column pruning*);
• choose the join strategy based on the estimated size of each side.

per_day.explain(mode="formatted")

explain shows the physical plan without running anything. Look for PushedFilters in the read and for Exchange (shuffle) between stages.

The trap: every action recomputes

Without a cache, two actions on the same DataFrame redo the whole plan, read included.

expensive = orders.join(customers, orders.customer_id == customers.id).where("amount > 100")
expensive.count()            # reads and joins
expensive.write.saveAsTable("gold.relevant_orders")  # reads and joins again

If the intermediate result is reused, write a table or use cache() (see Cache and persist).

• collect() brings everything to the driver: on a large table, it brings the driver down. Use take(n) or limit(n).
• print(df) runs nothing; it shows only the schema. df.show() is an action.
• Reading the schema can trigger a job if it has to be inferred (CSV/JSON without a declared schema).