Part I - How Spark Reads Parquet

chakshu_salgotra

7 min read

·

Jul 2, 2026

--

Listen

Share

Why a 540GB CSV brought a pipeline to its knees — and what columnar storage does about it

Part I

The 540GB CSV That Wouldn’t Die

The ask was innocent: ”Can you get me daily revenue from the events export?” The export was `events.csv` — a single 540GB file, gzipped, lovingly produced by a legacy system that had never heard of a data warehouse.

The query was one line: sum `amount` where `event_date = ‘2026–06–29’`. One column. One day. It ran for three hours and then died on an out-of-memory error.

Nothing was wrong with the query. Everything was wrong with the file. CSV is row-major — every value of every row is interleaved on disk, so to read one column Spark must read and parse every byte of all 540GB. There are no types (everything is a string until proven otherwise), no statistics to skip anything, and because the file was a single gzip stream it wasn’t even splittable— all 540GB funneled through one core while a 60-node cluster watched.

The fix wasn’t a bigger cluster. A senior engineer converted the file to Parquet once, re-ran the exact same query, and it came back in nine seconds.

So what is Parquet doing that CSV can’t — and, because this is data engineering and nothing is ever that easy, is simply switching formats actually enough?

Why CSV Chokes (and Columnar Doesn’t)

A CSV is a wall of text that happens to contain commas. Spark can read it, but every analytical query pays four taxes:

• Row-major layout — values are stored row by row, so reading one column means walking past every other column of every row. You cannot read `amount` without dragging `country`, `device`, and everything else along with it.
• No schema, no types — every field is text until parsed and cast. Type inference and string→number conversion burn CPU on data you may discard a millisecond later.
• No statistics — there is no `min`/`max` anywhere, so a `WHERE` clause can skip nothing. The engine reads everything, then throws most of it away.
• Compressed means non-splittable — a single gzipped CSV is one indivisible stream. Spark can’t hand slices to different tasks, so your parallelism collapses to one core.

Columnar formats invert all four. Here is the same data, both ways:

CSV row-major: 1 column = read all
row0: id ts amount country device …
row1: id ts amount country device …
row2: id ts amount country device …
^ sum(amount) scans every field
Parquet columnar: read only what you ask
[id] [ts] [amount]* [country] […]
^ read [amount]* only; skip rest

Converting is a one-time cost that pays out on every query afterward:

from pyspark.sql import functions as F
from pyspark.sql.functions import col
# NAIVE: query the 540GB CSV directly -> read and parse every byte of every
# row (and, gzipped, on a single task) just to sum one column.
csv = spark.read.option("header", True).csv("events.csv.gz")
csv.filter(col("event_date") == "2026–06–29").agg(F.sum("amount")).show()
# OPTIMIZED: pay the conversion once -> columnar, typed, compressed, with
# per-row-group stats and split into parallel-readable row groups.
csv.write.partitionBy("event_date").parquet("events_parquet")
# The same query now reads ~one column of one partition, not 540GB.
(spark.read.parquet("events_parquet")
.filter(col("event_date") == "2026–06–29")
.agg(F.sum("amount")).show())

Those nine seconds aren’t magic — they’re everything Parquet writes into the file around your data. And to see how that same machinery can still betray you, you have to open the file up.

A Parquet File Is Not a Table — It’s a Layout

Spark never “reads a table.” It opens files, reads each file’s footer, and decides — per chunk of rows — whether to touch the data at all. Those nine seconds — and every way Parquet can still disappoint you — come down to four nested structures.

Anatomy: files, row groups, column chunks, pages

• Row group — a horizontal slice of rows (default ~128MB). This is the unit of read parallelism and the unit of skipping. Spark either reads a row group or skips it whole.
• Column chunk — within a row group, all values for a single column stored contiguously. This is what makes Parquet columnar and enables reading one column without touching the others.
• Page— the smallest encoded/compressed unit inside a column chunk (data pages plus an optional dictionary page).
• Footer (`FileMetaData`) — written last, at the end of the file. It holds the schema, the byte offsets of every row group, and — critically — per-column-chunk statistics: min, max, and null_count. A reader seeks to the end, reads the PAR1 magic and a 4-byte footer length, then parses the footer before reading any data.

Parquet file
|
+- Row Group 0
| |- column chunk: user_id
| |- column chunk: amount
| +- column chunk: event_ts
|
+- Row Group 1
| +- …
|
+- Footer (FileMetaData)
|- schema
|- row-group offsets
+- per-row-group stats:
min / max / null_count

Why columnar wins twice

The columnar layout pays off in two independent ways:

• Projection pruning— `SELECT user_id, amount` reads exactly two column chunks per row group. A 40-column table costs you 2/40ths of the column I/O. `SELECT *` throws that away.
• Compression — values of one type sit together, so encodings get dramatic ratios. Dictionary encoding replaces repeated values with small integer codes; RLE/bit-packing crushes low-cardinality and sorted columns. Smaller pages mean fewer bytes off disk and faster decode.

Plot twist: the pushdown that skipped nothing

Months later the events feed had grown to a fat 220GB per day, all in Parquet — and a point lookup, `SELECT amount FROM events WHERE user_id = 738114`, still read an entire 220GB day-partition to return 3,000 rows. `explain()` showed `PushedFilters: [EqualTo(user_id, 738114)]`. The filter was pushed. Nothing was skipped. So much for columnar magic.

The reason exposes the most important rule about Parquet: a pushed-down filter only helps if the layout lets it.

The skip decision: statistics, not data

When Spark applies a pushed predicate, it does not decode rows to test them. It evaluates the predicate against the footer’s min/max statistics first (`spark.sql.parquet.filterPushdown`, default `true`):

If `738114` ∉ `[min_user_id, max_user_id]` for a row group, skip the entire row group - without reading or decompressing a single page.

Newer files carry a page index (a column index + offset index) that pushes the same logic down to individual pages, so even within a surviving row group Spark can skip most pages. This is predicate pushdown realized at the storage layer— and it is where the 220GB scan was lost.

Why the 220GB scan happened

Statistics only help when ranges are tight and non-overlapping. The `events` files were written unsorted on `user_id`. With rows in arrival order, every 128MB row group contained a near-random sample of the entire id space — so every row group’s `[min, max]` spanned almost the full range and overlapped the target:

Unsorted (ranges overlap -> skip nothing):
RG0 [ 1 .. 9,998,001 ]
RG1 [ 3 .. 9,999,400 ]
RG2 [ 2 .. 9,997,900 ]
id=738114 in every range -> read all
Clustered (ranges disjoint -> skip most):
RG0 [ 1 .. 120,000 ]
RG1 [120,001 .. 240,000 ]
…
RG6 [720,001 .. 840,000 ]
id=738114 -> RG6 only, rest skipped

Unsorted, the predicate matched every row group’s range, so Spark read all of them. The pushdown was real; the layout made it inert.

The fix: lay data out for the predicate

Cluster the data so row-group ranges partition the key space. In Spark, `repartitionByRange` assigns contiguous id ranges to files and `sortWithinPartitions` orders rows inside each file, producing tight, disjoint per-row-group stats:

• Result: after clustering on `user_id` with 128MB row groups, the same point lookup touched roughly 30 of ~1,700 row groups— 220GB → ~6GB, and the dashboard refresh dropped from minutes to seconds. Same query, same pushdown, different physical layout.

Two sizing traps

• Row group size. Too large → coarse skipping and memory pressure on read. Too small → footer/metadata overhead and weak compression. Stay in the 128MB–1GB band (`parquet.block.size`).
• The small-files problem. Millions of tiny files inflate the file index, multiply footer reads, and wreck parallelism. Compact them; don’t let a streaming sink emit 2KB Parquet files.

Reading and Fixing the Layout

You can inspect exactly what the reader sees. PyArrow exposes per-row-group statistics straight from the footer:

import pyarrow.parquet as pq
pf = pq.ParquetFile("events/event_date=2026–06–29/part-00000.parquet")
for i in range(pf.metadata.num_row_groups):
stats = pf.metadata.row_group(i).column(0).statistics # user_id
print(i, stats.min, stats.max) # overlapping ranges -> no skipping possible

Then fix the write so the ranges become disjoint:

# NAIVE: write in arrival order -> every row group's [min,max] spans the
# full user_id range -> a point-lookup filter can skip zero row groups.
df.write.parquet("events_unsorted")
# OPTIMIZED: cluster by the filter column so row-group ranges are tight and
# non-overlapping -> the same pushed filter now skips ~98% of row groups.
(df.repartitionByRange("user_id")
.sortWithinPartitions("user_id")
.write.parquet("events_clustered"))

And stop reading columns you never reference — projection pushdown is free, but only if you ask for it:

from pyspark.sql.functions import col
# NAIVE: SELECT * reads every column chunk in each surviving row group.
spark.read.parquet("events_clustered").filter(col("user_id") == 738114)
# OPTIMIZED: project first -> only the user_id and amount chunks are read.
(spark.read.parquet("events_clustered")
.select("user_id", "amount")
.filter(col("user_id") == 738114))

Takeaways

- CSV is honest and slow. A row-major, untyped, unindexed file forces Spark to read and parse every byte to answer a one-column question — and gzipped, it won’t even split across the cluster. For analytics, convert to columnar once and never look back.

- Parquet is a layout, not a table. Spark reads the footer first and decides, per row group, whether to touch the data — using `min`/`max`/`null_count` statistics, never the rows themselves.

- Pushdown is necessary, layout is sufficient. A pushed-down filter only skips bytes when row-group ranges are tight and non-overlapping. Cluster and sort on the columns you filter by.

- Project before you scan. Selecting only the columns you need turns a 40-column read into a 2-column read for free.

- Mind the granularity. 128MB–1GB row groups, and compact small files — both the floor and the ceiling cost you.

The storage layer can skip 98% of your data — but only if the query engine generates the right filters in the first place. How Spark decides what to push down, and how a single UDF can silently disable all of it, is [Part II — the Catalyst optimizer].

If this saved you a 3 AM pager incident, follow me on Medium for Twice a week for the next 3 Months of Advanced Data Engineering & AI. Let’s connect on LinkedIn and X.