Jobs That Run Before an Action

Concepts covered: Spark UI Diagnosis

The rule from the beginner lessons holds most of the time: transformations describe, actions run. It bends whenever Spark cannot finish describing the result without looking at data. A DataFrame must know every column's name and type from the line that creates it, and a few operations must know their output columns as well. When that information lives only inside the files, Spark launches a job on the spot, during the line you wrote.

Reading a file without a schema

CSV and JSON files carry no column types, so reading one without a schema forces Spark to open the data:

events = spark.read.csv("/data/events/", header=True, inferSchema=True)

That single line launches 2 jobs. The first reads the first line of the input, which gives the column count and, with header=True, the names. The second reads every row of every file to settle each column's type, because a single value of 3.5 at row 40 million turns an integer column into a double. On a 200GB directory that is a full scan, paid again when your first action reads the same files.

Declaring the schema removes the scan:

events = (spark.read
    .schema("event_id INT, user_id INT, event_type STRING, country STRING")
    .csv("/data/events/", header=True))

File listing still happens during the read. Spark lists the files on the driver, and when the directory tree holds more paths than spark.sql.sources.parallelPartitionDiscovery.threshold (32 by default), it runs the listing as a job of its own.

pivot without its values

A pivot turns the distinct values of 1 column into output columns. Spark cannot name those columns without the values, so this line runs a job:

by_status = orders.groupBy("region").pivot("status").sum("profit")

The job finds the distinct status values, sorts them and collects them to the driver. Past spark.sql.pivotMaxValues (10,000 by default) the line fails. Listing the values yourself skips the job and pins the output schema: a new status that shows up next month adds no surprise column for every downstream reader.

by_status = (orders.groupBy("region")
    .pivot("status", ["Completed", "Pending", "Cancelled"])
    .sum("profit"))

Extra jobs hiding inside one action

Some actions launch more than 1 job before the result arrives. A sort is the common case: orderBy sends rows to tasks by range of the sort key, and to choose the range boundaries Spark first runs a job that samples the key, then runs the work itself. show() is the other: it runs a job over 1 partition, and if that produces fewer than the 20 rows it wants, it runs another over up to 4 times as many partitions, and again, until it has enough. After a filter that keeps 0.01% of the rows, 1 show() can launch several jobs, each scanning more new partitions than the last.

Which lines launch a job before any action runs?

events = spark.read.json("/data/events/")
typed = spark.read.schema("user_id INT, event_type STRING, country STRING").json("/data/events/")
counts = events.groupBy("country").pivot("event_type", ["click", "view"]).count()
wide = typed.groupBy("country").pivot("event_type").count()

• Only the wide line, since pivot is the only operation here that must look at data
• The events read and the counts pivot, since every pivot collects its distinct values
• None of them, since Spark opens no file until an action such as collect() runs
• The events read and the wide pivot: one lacks a schema, the other its output columns

What the value list decides

A value list skips the job, and it also fixes which columns the result has. A status that appears in the data but is missing from the list is dropped from the output, and its profit is counted in no column. A listed status that never occurs still gets a column, filled with null for every region. The list is part of the output contract: when a new status starts to matter, adding it is a deliberate change made in 1 place, and every reader of the table sees the new column on the same day.

The orders data holds the statuses Completed, Pending and Refunded, and no row is Cancelled. Which columns does by_status have, and where does the Refunded profit go?

by_status = (orders.groupBy("region")
    .pivot("status", ["Completed", "Pending", "Cancelled"])
    .sum("profit"))

• region, Completed, Pending and Refunded: Spark adds a column for each status the data holds
• region, Completed, Pending and Cancelled; Cancelled is null everywhere and Refunded profit is dropped
• region, Completed and Pending; Cancelled is left out because no row in the data holds it
• region, Completed, Pending, Cancelled and Refunded, with Cancelled null for every region

About this lesson

This section is part of the Lazy Evaluation: Intermediate 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.