Module 3: Rebuilding Fact Orders With The Dataframe Api

Writing `fact_orders` as Parquet

Next →

Description

Up to this lesson, fact_orders_df only existed as an in-memory execution plan — read, joined, calculated, aggregated, verified, but never saved. This lesson writes it, for the first time in this entire guide, as a real file on disk: fact_orders.parquet. It isn't just any file: it's the same artifact lakehouse-and-iceberg-guide, the sister guide that follows this one in the ecosystem, is going to reuse to build native versioning and schema evolution on top of.

Connection to the module. This lesson takes lesson 6's already-verified fact_orders_df — forty rows, 106.15 in total revenue, confirmed against three previous engines — and persists it. Everything that follows in the module (lesson 8's project) assumes this file already exists on disk, not just in memory.

An analogy: filing the final card in the right drawer

Pick back up, one last time in this module, the card archive. You already assembled the complete card (lesson 3), added the subtotal (lesson 4), grouped it and confirmed the piles add up correctly (lessons 5 and 6). All that work, so far, lived on the desktop — loose papers, useful while you work, but that vanish the moment the office closes. Writing fact_orders.parquet is filing those finished cards into a drawer with its own label, ready for anyone else — or any other guide in this same ecosystem — to open later without having to redo the assembly work from scratch.

Worked example: .write.parquet() and rereading

Step 1 — Pick back up fact_orders_df, already verified

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

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

orders_schema = StructType([
    StructField("order_id", StringType(), 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_df = spark.read.csv("orders_2026-08-*.csv", schema=orders_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_df = (
    orders_df
    .join(dim_store_df, "store_id")
    .join(dim_product_df, "product_id")
    .withColumn("revenue", col("quantity") * col("unit_price"))
    .select("order_id", "store_id", "product_id", "quantity", "unit_price", "revenue", "order_ts")
)

Notice the final .select(): it reorders and trims the columns down to the seven that actually matter for fact_orders — leaving out store_name, city, product_name, category, unit_cost, which are dimension attributes, not fact attributes. fact_orders keeps the keys (store_id, product_id) to rejoin against the dimensions whenever needed, not a copy of their attributes.

Step 2 — Write as Parquet

fact_orders_df.write.mode("overwrite").parquet("fact_orders.parquet")
print("Written: fact_orders.parquet")

What to expect. Running this part, there's no screen output beyond the print — but a new folder shows up on disk:

Written: fact_orders.parquet

$ ls fact_orders.parquet/
_SUCCESS
part-00000-....snappy.parquet
part-00001-....snappy.parquet
part-00002-....snappy.parquet
part-00003-....snappy.parquet
part-00004-....snappy.parquet
part-00005-....snappy.parquet
part-00006-....snappy.parquet

fact_orders.parquet is not a single file — it's a folder with seven .parquet files (plus an empty _SUCCESS marking the write finished with no errors). Seven, the exact same number of partitions orders_df carried since the original read — one for each of the seven CSV files for Kiosko's week: since lesson 3's two JOINs didn't change that partition count (a Spark decision you're going to fully understand in module 5, when you learn about broadcast joins), the write keeps the same count. The seven files' total size, executed in this run, is 15,721 bytes — about 15 KB for forty rows — an honest reminder that Parquet carries metadata overhead (schema, per-column statistics, snappy compression) that only pays off with real volume; over forty rows, a flat CSV would weigh less. Parquet's advantage doesn't show up at this scale — it shows up in module 4, when the same format has to carry ten million rows.

Step 3 — Reread and verify

reread_df = spark.read.parquet("fact_orders.parquet")
print(f"reread_df.count() = {reread_df.count()}")

reread_total = reread_df.agg(F.round(F.sum("revenue"), 2).alias("total")).collect()[0]["total"]
print(f"reread_total = {reread_total}")

assert reread_df.count() == 40
assert reread_total == 106.15
print("Verification: fact_orders.parquet reread == 40 rows, total == 106.15 -> OK")

reread_df.printSchema()

spark.stop()

What to expect (executed in this run):

reread_df.count() = 40
reread_total = 106.15
Verification: fact_orders.parquet reread == 40 rows, total == 106.15 -> OK
root
 |-- order_id: string (nullable = true)
 |-- store_id: string (nullable = true)
 |-- product_id: string (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- unit_price: double (nullable = true)
 |-- revenue: double (nullable = true)
 |-- order_ts: timestamp (nullable = true)

The schema survives the write and the reread completely — quantity is still integer, unit_price and revenue are still double, order_ts is still timestamp — without you having to declare any StructType to read the Parquet back. This is a real difference from module 1's CSV: Parquet stores the schema inside the file itself, as part of its metadata, so spark.read.parquet(...) doesn't need you to tell it the types — it already knows them.

Diagram: from memory to disk, and back

flowchart LR
    A["fact_orders_df\n(in memory, verified plan)"] -->|".write.mode('overwrite')\n.parquet(...)"| B["fact_orders.parquet/\n7 .parquet files + _SUCCESS\n~15 KB"]
    B -->|"spark.read.parquet(...)"| C["reread_df\n40 rows, same schema,\nno StructType declared"]
    C -.->|"assert count==40,\ntotal==106.15"| D(("Verified"))

Going deeper: why fact_orders doesn't store store_name or product_name

Look again at Step 1's .select(): fact_orders.parquet stores store_id and product_id — the keys — but not store_name, city, product_name, category, or unit_cost — the dimension attributes. This isn't an oversight; it's, precisely, the discipline of a star schema data-modeling-for-analytics-guide already justified: the fact table stores foreign keys and metrics (quantity, unit_price, revenue), and anyone who needs a store's or product's readable name rejoins against dim_store or dim_product at query time, instead of loading that duplicated information into every row of fact_orders.

This has a real practical consequence, worth seeing with a number: if dim_store changed — say, if Kiosko Norte got renamed to something else — a fact_orders that had stored store_name as its own column would have to be rewritten row by row to reflect the change. A fact_orders that only stores store_id doesn't need to be touched at all: the next time someone joins it against dim_store, they automatically see the updated name. Storing only the keys, not the attributes, is what keeps a change in a dimension from forcing a rewrite of the entire fact history — one of the underlying reasons, though this guide doesn't build it out in depth here, why data-modeling-for-analytics-guide devoted a full module to slowly changing dimensions (SCD).

Common mistakes

Writing without .mode("overwrite"), and running into an error the second time you run the script. What happens: someone runs fact_orders_df.write.parquet("fact_orders.parquet") (without .mode(...)) once, and everything works — but running the same script a second time, without deleting the folder by hand first, the script fails. Here it is, executed, exactly that error:

df = spark.read.parquet("fact_orders.parquet")
df.write.parquet("fact_orders.parquet")  # no mode(), and the folder already exists

What to expect (executed in this run, abbreviated message):

AnalysisException: [PATH_ALREADY_EXISTS] Path file:.../fact_orders.parquet already exists.
Set mode as "overwrite" to overwrite the existing path. SQLSTATE: 42K04

Why it happens: .write's default mode in Spark is error (also called errorifexists) — a deliberate design decision so you never overwrite data by accident without explicitly asking for it. How to spot it: if your script fails with PATH_ALREADY_EXISTS the second time you run it (but not the first), you're almost certainly missing .mode("overwrite"). How to fix it: for a script that rebuilds fact_orders from scratch every time it runs — like this lesson's — .mode("overwrite") is the right choice; for a pipeline that appends new data to an existing history without erasing what came before, .mode("append") would be the right choice instead — a different design decision this guide doesn't need yet.

Expecting a single .parquet file, not a folder with several. What happens: someone looks for a file literally named fact_orders.parquet (a single binary file) and is confused to find a folder with that name, full of part-*.parquet files. Why it happens: in single-node tools, like pandas (df.to_parquet("file.parquet")), the result usually is a single file. How to spot it: if your code in another tool expects to open fact_orders.parquet as an individual file and fails because it's actually a directory, check that you're using spark.read.parquet("fact_orders.parquet") (which knows how to read the whole folder as a single table), not a function expecting a single file. How to fix it: this is, in fact, a direct and correct consequence of Spark's distributed model — each partition (each executor, working in parallel) writes its own file, and all of them together form the complete table; you never need to merge them by hand, spark.read.parquet() treats them as a unit automatically.

Not verifying the result after rereading, trusting that "if there was no error, it's fine." What happens: someone writes the Parquet, sees no error on screen, and assumes the content is correct without rereading it and counting the rows. Why it happens: the absence of a write error feels like sufficient confirmation. How to spot it: the absence of an error only confirms Spark managed to write something to disk — it doesn't confirm that something has the correct row count or the correct revenue total. How to fix it: this lesson's Step 3 — reread and verify again with assert — isn't optional; it's the same "verify, don't assume" discipline that runs through this whole module, now applied to the final artifact on disk, not just the in-memory DataFrame.

Exercises

Exercise 1 — Confirm the Parquet on disk weighs less than the equivalent CSV, at a different data scale. Compare, with os.path.getsize(), the total size of the seven orders_2026-08-*.csv files (from module 1) against the total size of fact_orders.parquet (more columns, same row count). Which one is heavier, and why would that make sense, even though fact_orders.parquet has more columns?

See solution 
import glob
import os

csv_bytes = sum(os.path.getsize(f) for f in glob.glob("orders_2026-08-*.csv"))
parquet_bytes = sum(
    os.path.getsize(f) for f in glob.glob("fact_orders.parquet/part-*.parquet")
)
print(f"Total CSV (orders, 6 columns): {csv_bytes} bytes")
print(f"Total Parquet (fact_orders, 7 columns): {parquet_bytes} bytes")

Expected output (Parquet's exact values may vary slightly between runs due to the UUID in the filename, but the order of magnitude is stable):

Total CSV (orders, 6 columns): 2206 bytes
Total Parquet (fact_orders, 7 columns): 15721 bytes

Here Parquet weighs more, not less, despite snappy compression — and the reason is honest, not an awkward surprise: at forty rows, Parquet's fixed metadata overhead (schema, per-column statistics, headers for each of the seven files) weighs more than the few bytes of real data there are to compress. Parquet's size advantage over CSV shows up with volume — tens of thousands of rows onward, not at forty — exactly the same point this lesson already made in "going deeper."

Exercise 2 — Write fact_orders filtered to just S01, in a separate folder, and verify its count. Using .filter(col("store_id") == "S01") before .write.parquet(...), write a separate Parquet (fact_orders_s01.parquet) and confirm it has sixteen rows.

See solution 
fact_orders_df.filter(col("store_id") == "S01").write.mode("overwrite").parquet("fact_orders_s01.parquet")

s01_df = spark.read.parquet("fact_orders_s01.parquet")
print(f"s01_df.count() = {s01_df.count()}")
assert s01_df.count() == 16
print("Verification: fact_orders_s01.parquet has 16 rows -> OK")

Expected output:

s01_df.count() = 16
Verification: fact_orders_s01.parquet has 16 rows -> OK

Sixteen, the same number of S01 orders you already know from the module 1 project. This pattern — filtering before writing, instead of writing everything and filtering after — is a preview of what module 7 is going to formalize with partitionBy("store_id"): instead of writing a separate Parquet per store by hand, Spark can automatically organize files into subfolders by a column's value.

Exercise 3 — Explain, without code, why fact_orders.parquet doesn't store store_name or product_name. In 2-3 sentences, and using the "going deeper" section's Kiosko Norte rename example, explain what problem storing only store_id avoids, instead of storing store_name directly in fact_orders.

See solution

If fact_orders stored store_name as a direct copy in every row, any future change to a store's name in dim_store — a rename, a spelling correction — would leave fact_orders's history out of date, with the old name frozen in every already-written row, unless the entire history got manually rewritten. By storing only the key (store_id), fact_orders never needs to be touched when dim_store changes: the next time someone joins the two tables, they automatically get the updated name, with no extra work at all. This is the practical, not just theoretical, reason behind separating facts and dimensions into different tables.

Summary and next step

In this lesson you wrote fact_orders_df — already verified in lesson 6 — as a real Parquet file on disk, with .write.mode("overwrite").parquet(...), and confirmed it read back correctly: forty rows, 106.15 in total revenue, the same typed schema with no need to declare a StructType again. You understood why fact_orders.parquet is a folder with several files, not a single one, and why it stores only the dimension keys, not their attributes.

Before moving on you should be able to: write the .write.mode("overwrite").parquet(...) line from memory; explain why Parquet at this scale weighs more than the equivalent CSV; and explain why fact_orders doesn't duplicate store_name or product_name inside its own rows.

Lesson 8 — this module's closing project — pulls the previous six lessons together into a single script verified end to end, from reading the three tables through fact_orders.parquet written and confirmed on disk.

Resources

← PreviousVerifying the same 106.15 totalNext →Project: Kiosko's `fact_orders` in Spark