Skip to main content

Module 4 — Reading data and columnar formats

Nothing said so far has cost anything, because no data has moved. This module is where the flight-delays project actually starts: we point Spark at tens of millions of rows and decide, before writing a single feature, how they are laid out on disk. That decision does more for performance than anything else in the course.

CSV is a trap at this scale​

The public flights dataset ships as a set of CSV files, one per year, totalling around 50 GB and roughly 200 million rows. The naive read looks like this:

flights = spark.read.option("header", True).csv("s3://open-data/flights/*.csv")
flights.count()

Three things go wrong at once. First, Spark infers the schema by scanning the whole dataset: it reads every row, twice, before it even starts computing anything else. On 50 GB this is a ten-minute tax on every job. Second, all columns are typed as string if you skip inference, forcing every downstream cast. Third, and worst, CSV is row-oriented: selecting three columns still reads all thirty from disk, because there is no way to skip the others.

Parquet solves all three problems​

Parquet stores columns separately on disk, with row-group statistics (min, max, count) written alongside the data. This makes it possible to read only the columns you need, and to skip entire row groups that cannot match a filter.

schema = "carrier STRING, origin STRING, dest STRING, distance INT, " \
         "departure_delay INT, arrival_delay INT, cancelled BOOLEAN, " \
         "year INT, month INT, day INT"

flights = (
    spark.read
    .schema(schema)
    .parquet("s3://open-data/flights-parquet/")
)

On the same data as before, this read finishes in seconds because no rows are scanned upfront: Spark reads the Parquet footer, gets the schema and statistics for free, and defers everything else until an action fires.

Two rules from this snippet deserve to be spelled out. Provide an explicit schema every time — even on Parquet, this saves a metadata call per file. And prefer Parquet over CSV for anything you own end to end; the conversion pays for itself in the first job.

Partitioned layouts​

A Parquet dataset written by year and month lays itself out on disk as a folder tree:

flights-parquet/
  year=2018/month=01/part-00000.parquet
  year=2018/month=02/part-00000.parquet
  ...
  year=2024/month=12/part-00000.parquet

Spark reads the partition values from the folder names — they never sit inside the files themselves. A query with a filter on year and month then benefits from partition pruning: only the folders that match the filter are opened.

january_2024 = flights.filter((F.col("year") == 2024) & (F.col("month") == 1))
january_2024.explain()
# ... PartitionFilters: [year=2024, month=1], PushedFilters: []

The PartitionFilters line in the plan tells you the pruning worked: one folder was opened, not eighty-four. Get the partitioning right and a "year of data" query takes two seconds instead of two minutes.

Choose partition columns carefully. Good candidates have a small number of distinct values (dozens to thousands) and appear in most WHERE clauses. Bad candidates are high-cardinality columns like a flight identifier or a timestamp with second precision: partitioning on them creates millions of tiny files that hurt every subsequent read.

Aligning partitioning with the shuffle​

The partitioning of the files on disk is not the same as the partitioning of a DataFrame at runtime. A Parquet read produces one Spark partition per file (or per group of small files, if spark.sql.files.maxPartitionBytes is respected). Once the data is in memory, groupBy(carrier) will reshuffle it into spark.sql.shuffle.partitions partitions.

The two do interact, though. If the disk is partitioned by carrier and your first operation is groupBy(carrier), Spark can sometimes avoid the shuffle entirely — the executor that reads a carrier's file is the one that will aggregate it. Aligning the disk layout with the most frequent grouping is one of the few tuning tricks that costs nothing at query time.

One-time conversion, permanent gain

On the flight-delays project, converting the CSV archive to Parquet partitioned by year is a five-minute batch job that runs once. It cuts every subsequent training job by an order of magnitude, and it costs the same disk space thanks to Parquet's built-in Snappy compression.

Summary​

  • CSV forces schema inference and reads every column of every row; avoid it at scale.
  • Parquet stores columns separately with row-group statistics, enabling projection and predicate pushdown.
  • Always pass an explicit schema at read time to skip the metadata scan entirely.
  • Partitioned layouts allow partition pruning; pick low-cardinality columns that appear in most filters.

Next module: turning those rows into features Spark ML can consume — transformers and estimators.