Parquet at scale: partitioned writes

Description

Module 3 wrote fact_orders.parquet just once, over forty rows, as a flat file — with no internal organization beyond its seven part-*.parquet files, one per in-memory partition inherited from the original read. This lesson picks back up exactly fact_orders_at_scale_df — the ten million rows module 6 rebuilt with two broadcast joins and cached for three queries — and writes it for the first time with partitionBy("store_id"): instead of a folder with loose files, three labeled subfolders, one per store, each containing only that store's data.

Connection to the module. This lesson is this whole module's hinge: without a partitioned Parquet on disk, lesson 3 has nothing to prune, and lesson 8 has nothing to run the pandas_udf on at full scale. Everything that follows in this module assumes fact_orders_at_scale.parquet already exists, partitioned, on disk.

An analogy: the warehouse that organizes when storing, not when searching

Pick back up this guide's module 1 analogy: the boxes split across trucks are in-memory partitions. This lesson builds something different but related: a physical warehouse, on disk, with three labeled aisles — store_id=S01, store_id=S02, store_id=S03 — built at the exact moment the boxes get stored, not when someone goes looking for them. If you stored Kiosko's ten million boxes with no order at all, any future search by store would have to check the entire warehouse, box by box. If you store them already separated by aisle — the work partitionBy("store_id") does in this lesson — that organization cost gets paid once, on write, and collected every time someone searches for just one store — lesson 3's exact topic.

Worked example: rebuild, and write partitioned

Step 1 — Pick back up fact_orders_at_scale_df, just like in module 6's project

You need kiosko_orders_at_scale.csv (module 4, 10,000,000 rows), dim_store.csv, and dim_product.csv (module 3), in the same folder.

# write_partitioned_fact_orders_at_scale.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").config("spark.driver.memory", "3g").getOrCreate()

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
dim_store_schema = StructType([
    StructField("store_id", StringType(), False),
    StructField("store_name", StringType(), False),
    StructField("city", StringType(), False),
])
dim_product_schema = StructType([
    StructField("product_id", StringType(), False),
    StructField("product_name", StringType(), False),
    StructField("category", StringType(), False),
    StructField("unit_cost", DoubleType(), False),
])

orders_at_scale_df = spark.read.csv("kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False)
dim_store_df = spark.read.csv("dim_store.csv", schema=dim_store_schema, header=True, enforceSchema=False)
dim_product_df = spark.read.csv("dim_product.csv", schema=dim_product_schema, header=True, enforceSchema=False)

fact_orders_at_scale_df = (
    orders_at_scale_df
    .join(dim_store_df, "store_id")
    .join(dim_product_df, "product_id")
    .withColumn("revenue", F.round(F.col("quantity") * F.col("unit_price"), 2))
)

print(f"fact_orders_at_scale_df.count() = {fact_orders_at_scale_df.count()}")
print(f"columns ({len(fact_orders_at_scale_df.columns)}) = {fact_orders_at_scale_df.columns}")

So far, nothing new: it's exactly the same thirteen-column fact_orders_at_scale_df module 6 already cached and queried three times.

Step 2 — Write with partitionBy("store_id")

fact_orders_at_scale_df.write.mode("overwrite").partitionBy("store_id").parquet("fact_orders_at_scale.parquet")
print("Written: fact_orders_at_scale.parquet, partitioned by store_id")

spark.stop()

What to expect. Running python3 write_partitioned_fact_orders_at_scale.py, the output is exactly this (executed in this run, PySpark 4.2.0):

fact_orders_at_scale_df.count() = 10000000
columns (13) = ['product_id', 'store_id', 'order_id', 'franchise_id', 'quantity', 'unit_price', 'order_ts', 'store_name', 'city', 'product_name', 'category', 'unit_cost', 'revenue']
Written: fact_orders_at_scale.parquet, partitioned by store_id

And on disk, a structure completely different from module 3's fact_orders.parquet:

$ find fact_orders_at_scale.parquet -maxdepth 1
fact_orders_at_scale.parquet
fact_orders_at_scale.parquet/_SUCCESS
fact_orders_at_scale.parquet/store_id=S01
fact_orders_at_scale.parquet/store_id=S02
fact_orders_at_scale.parquet/store_id=S03

$ ls fact_orders_at_scale.parquet/store_id=S01/ | head -3
part-00000-30675c7c-....c000.snappy.parquet
part-00001-30675c7c-....c000.snappy.parquet
part-00002-30675c7c-....c000.snappy.parquet

Instead of a single folder with loose files — what you saw with fact_orders.parquet in module 3 — fact_orders_at_scale.parquet is a folder with three subfolders, each with the literal name store_id=<value> — this is what Spark's official documentation calls partition discovery: the partition column's value gets encoded in the folder name, not inside the .parquet files. Notice something important: store_id no longer shows up as just another column inside each file's data — it moved "outside," into the directory name.

The real sizes, measured in this run:

Three things to verify by hand. First, every partition has 12 files — the same in-memory partition count (numCachedPartitions=12) you already saw in module 6's project for this same DataFrame before writing it; partitioning by store_id on disk doesn't change how many in-memory partitions the DataFrame had at write time, each in-memory partition simply gets spread across the destination folders. Second, S01 weighs more than S02, which in turn weighs more than S03 — the same row proportion you already know from module 1 (S01=16, S02=13, S03=11 orders per franchise, scaled ×250,000: S01=4,000,000, S02=3,250,000, S03=2,750,000 rows). Third, and most important: 56,639,485 bytes total, against kiosko_orders_at_scale.csv's 601,305,672 bytes — the source CSV, with only seven columns. Even though fact_orders_at_scale.parquet has nearly double the columns (thirteen, from the two joins), it weighs 10.6 times less than the source CSV. This is exactly Parquet's advantage module 3 couldn't show with forty rows — there, Parquet weighed more than the CSV, because of Parquet's fixed metadata overhead — at ten million rows, snappy columnar compression, plus the fact columns like store_name, city, product_name, and category only have a handful of distinct values repeated ten million times, makes Parquet win by a huge margin.

Diagram: from a flat folder to a warehouse with three aisles

flowchart TD
    A["fact_orders_at_scale_df\n10,000,000 rows, 13 columns\n(in memory, already cached in M6)"] -->|".write.mode('overwrite')\n.partitionBy('store_id')\n.parquet(...)"| B

    subgraph B["fact_orders_at_scale.parquet/"]
        C["store_id=S01/\n12 files, 21,721,531 bytes\n4,000,000 rows"]
        D["store_id=S02/\n12 files, 18,775,375 bytes\n3,250,000 rows"]
        E["store_id=S03/\n12 files, 16,142,355 bytes\n2,750,000 rows"]
    end

    B -.->|"total: 56,639,485 bytes\n~10.6x lighter than the CSV (601,305,672 bytes)"| F(("Lesson 3:\npartition pruning"))

Going deeper: which column to partition by, and why store_id

Choosing the partitionBy(...) column isn't a decision free of costs in every direction: the practical rule, confirmed by Spark's official documentation on partition discovery, is to partition by a low-cardinality column — few distinct values — that also matches your future queries' most frequent filter pattern. store_id meets both conditions in Kiosko: it only has three possible values (S01, S02, S03), and this guide's module 5 already established "give me one store's data" as exactly the kind of filter Kiosko needs — it's the same column you already used for the broadcast join against dim_store.

It's worth reasoning through, without running it, what would happen if you partitioned by franchise_id instead — this dataset's synthetic column, with 250,000 distinct values. partitionBy("franchise_id") would produce, in theory, up to 250,000 subfolders — one per franchise — each with a handful of tiny files (forty rows per franchise, split across a few in-memory partitions). This is known as the "small files problem": the overhead of opening, listing, and reading the metadata of tens of thousands of tiny files ends up outweighing partition pruning's benefit. The practical rule this lesson leaves you with: partition by a column with few distinct values and a real query pattern that uses it — never by the column with the most distinct values "because it looks more specific."

Common mistakes

Writing without .mode("overwrite"), expecting the same behavior as module 3. What happens: someone runs this script a second time, without .mode("overwrite"), and hits the same PATH_ALREADY_EXISTS error module 3 already documented. Why it happens: it's .write's same default behavior (mode="error"), with no relation to partitioning. How to spot it: the message is identical to module 3's, just now pointing at fact_orders_at_scale.parquet. How to fix it: the same discipline as always — .mode("overwrite") for a script rebuilding the Parquet from scratch every time, like this lesson's.

Confusing store_id as a DataFrame column after reading the partitioned Parquet back. What happens: someone, after reading spark.read.parquet("fact_orders_at_scale.parquet"), does .select("store_id") and is surprised it works — they expected that, since store_id "moved" into the folder name, it would no longer be available as a column. Why it happens: it's easy to think partitioning "removes" the column from the visible schema. How to spot it: check printSchema() on the reread DataFrame — store_id still shows up, with its correct type (string), exactly like the other twelve columns. How to fix it: Spark rebuilds the partition column automatically from the folder name when reading with spark.read.parquet(...) over the root folder — the column never disappears from the logical DataFrame, it's just stored differently on disk (as part of the directory name, not inside each individual .parquet file). Lesson 3 shows exactly this mechanic in the .explain() plan.

Partitioning by a high-cardinality column "for more control," without measuring the small-files cost. What happens: someone, excited by partitionBy("store_id")'s benefit, decides to partition their own dataset by a column with thousands or hundreds of thousands of distinct values — a second-granularity date, a user ID. Why it happens: "more partitions" intuitively sounds like "more organized," without considering the overhead of listing and opening a huge number of tiny files. How to spot it: if your partitioned Parquet ends up with more folders than rows per folder, or with files only a few kilobytes each, that's a clear sign of the small-files problem. How to fix it: this lesson's "going deeper" section already reasoned through it with franchise_id (250,000 values) without needing to run it — partition by low-cardinality columns, aligned with the real query pattern, never by the most specific column available.

Exercises

Exercise 1 — Verify, with printSchema(), that store_id is still in the schema after rereading the partitioned Parquet. Read fact_orders_at_scale.parquet back with spark.read.parquet(...) and confirm all thirteen columns — including store_id — are still present, with the same types.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

reread_df = spark.read.parquet("fact_orders_at_scale.parquet")
reread_df.printSchema()
print(f"reread_df.count() = {reread_df.count()}")
assert reread_df.count() == 10_000_000
assert "store_id" in reread_df.columns
print("Verification: store_id is still in the schema, 10,000,000 rows -> OK")

spark.stop()

root
 |-- product_id: string (nullable = true)
 |-- order_id: string (nullable = true)
 |-- franchise_id: integer (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- unit_price: double (nullable = true)
 |-- order_ts: timestamp (nullable = true)
 |-- store_name: string (nullable = true)
 |-- city: string (nullable = true)
 |-- product_name: string (nullable = true)
 |-- category: string (nullable = true)
 |-- unit_cost: double (nullable = true)
 |-- revenue: double (nullable = true)
 |-- store_id: string (nullable = true)
reread_df.count() = 10000000
Verification: store_id is still in the schema, 10,000,000 rows -> OK

Exercise 2 — Calculate, without running Spark, how many rows store_id=S02 should have after writing the partitioned Parquet, using the proportion you already know from module 1. Module 1 established S02 has 13 orders in every real Kiosko week (out of a total of 40). With 250,000 franchises, how many rows land in store_id=S02/?

Exercise 3 — Explain, without code, why fact_orders_at_scale.parquet weighs less than kiosko_orders_at_scale.csv, despite having nearly double the columns. In 2-3 sentences, and using this lesson's real numbers (56,639,485 bytes against 601,305,672 bytes), explain which two Parquet factors explain that difference.

Summary and next step

In this lesson you wrote, for the first time in this guide, a partitioned Parquet: fact_orders_at_scale.parquet, 10,000,000 rows spread across three subfolders labeled by store_id, with partitionBy("store_id"). You confirmed, with real numbers, that the result weighs 56,639,485 bytes — 10.6 times less than the source CSV, despite having more columns — and understood why store_id is a good partition column for Kiosko: low cardinality, aligned with the real query pattern.

Before moving on you should be able to: write the line .write.mode("overwrite").partitionBy("store_id").parquet(...) from memory; explain the difference between how a partition column gets stored (in the folder name) and a normal column (inside the file); and explain, in your own words, why partitioning by franchise_id would be a mistake in this dataset.

Lesson 3 takes this same partitioned Parquet and measures, with .explain(), exactly how much work Spark saves when a query filters by the partition column — and, by contrast, how much it saves when it filters by a column that isn't one.

Resources

• Apache Spark — SQL Data Sources: Parquet (partition discovery, the column=value folder-name encoding, and spark.sql.parquet.filterPushdown). spark.apache.org/docs/latest/sql-data-sources-parquet.html.
• Apache Spark — DataFrameWriter.partitionBy (the exact API reference used in this lesson). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.partitionBy.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — the section declaring partitioned fact_orders_at_scale.parquet as the artifact lakehouse-and-iceberg-guide is going to reuse to build native time travel and schema evolution.