The Action Catalog
Concepts covered: Spark Execution Model
Every action starts a job, and the actions differ in what they hand back, in how much of the chain they have to run, and in where the result lands. Knowing the catalog tells you what a line will cost before you run it.
| Action | Returns | How much it runs | Where the result lands |
|---|---|---|---|
| `count()` | 1 number | Every partition | Driver, as an integer |
| `collect()` | Every row | Every partition | Driver memory, as a list of Row |
| `show(n)` | Prints n rows, 20 by default | Partitions until it has n rows | Driver, printed |
| `take(n)`, `head(n)`, `first()` | n rows, or 1 | Partitions until it has n rows | Driver, as a list of Row |
| `write.parquet(path)` | Nothing | Every partition | Files at the path, nothing on the driver |
| `toPandas()` | Every row | Every partition | Driver memory, as one local table |
Actions that stop early
show() and take() ask for a handful of rows, and Spark exploits that. For a chain of per-row steps, a filter followed by a select, a task can produce a finished row as soon as it reads one. Spark starts a job over 1 partition, and if that partition yields 20 rows the job ends there. A show() on a 400-partition table reads 1 partition in a second, while a count() on the same table reads all 400. If the first partition yields fewer than 20 rows, Spark launches another job, and each new round tries about 4 times as many partitions as it has scanned so far, until it has enough rows.
Stopping early needs finished rows
The shortcut works only when a finished row can exist before the whole chain finishes. After a groupBy, no result row is complete until every source row for that key has been seen, so show() on a grouped DataFrame runs the full grouping and then prints 20 of the results. The same holds for orderBy: the first row of a sorted result is known only after every partition has been read. A quick look is cheap on the chain before the grouping, where it costs 1 partition, and costs the whole job after it.
Lines that start no job
Each of these runs in milliseconds and sends no task
- printSchema() reads the column list that arrived with the file listing.
- columns, dtypes and schema read the same metadata.
- limit(20) records a cap on the rows and waits for an action.
- distinct(), orderBy() and groupBy().agg() record steps that will run later, however expensive they are.
- df.write on its own is a builder, and the job starts at .parquet(path) or .save().
The Jobs page of the Spark UI lists the jobs each action started, which makes the catalog visible after the fact. Every action starts at least 1 job: a pipeline with 1 write usually shows 1, and a notebook cell with 3 show() calls shows 3 or more, because a show() whose first partition comes up short adds follow-up jobs. A line that appears there cost a job, whatever the code looked like.
How many lines in the program below start a job?
events = spark.read.parquet("s3://logs/events/")
errors = events.filter(F.col("level") == "ERROR")
errors.printSchema()
sample = errors.limit(10)
sample.show()
n = errors.count()- Two, the show() line and the count() line, while the other lines record steps.
- Three, because printSchema() also has to read the files to learn the columns.
- Four, one for each line after the read, since each one touches the data.
- One, because Spark folds every line of the cell into a single job.
Picking the action is picking the cost. A look at 5 rows of a filtered table needs an action that stops early. show(5) and take(5) both do, and either one usually reads 1 partition. The number of rows the filter keeps can come only from an action that runs every partition, and count() does that while sending the driver 1 number.
The table trips is stored as 400 partitions, and completed is a filter on it with plenty of matching rows in every partition. Which line usually reads the fewest partitions?
- completed.count(), because the driver receives only 1 number from the job.
- completed.groupBy("city").count().show(5), because only 5 grouped rows get printed.
- completed.show(5), because a filtered row is finished as soon as a task reads it.
- completed.orderBy("fare_cents").show(5), because only the 5 lowest fares travel.
About this lesson
This section is part of the Lazy Evaluation: 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.