Project: Kiosko's `fact_orders` in Spark

Description

This project closes the module by pulling the previous six lessons together into a single delivery: you read the three tables with an explicit schema, join them with .join(), calculate revenue with .withColumn(), group by store and by product with .groupBy().agg(), verify the result against the ecosystem's four previous engines with assert, and write fact_orders.parquet to disk — the artifact the rest of this guide, and lakehouse-and-iceberg-guide later in the ecosystem, are going to reuse.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of lessons 2 through 7, in the same order you built them, with a single script verified end to end.

An analogy: the fourth balance, delivered in full

Pick back up, one last time in this module, with lesson 1's four accountants. This project is the moment the fourth accountant — the one who learned Spark — turns in their complete balance: not just the final number, but the entire process that produced it, documented and verifiable from the first raw piece of data through to the final filed record. A balance that only shows the result, without the verified path that produced it, doesn't inspire the same confidence as one that shows every step, with every intermediate check passed.

The material: everything this module built, in a single flow

You need: the seven orders_2026-08-03.csv through orders_2026-08-09.csv files from module 1, and this module's two new files from lesson 2 (dim_store.csv, dim_product.csv), all in the same folder where you're going to run the script.

The reference solution, verified

Parts 1 and 2 — Open the session, read the three tables

# kiosko_fact_orders_in_spark.py
import glob

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,
)

print("=== Kiosko in Spark: fact_orders rebuilt, module 3 final delivery ===\n")

print("Part 1 -- opening the SparkSession")
spark = (
    SparkSession.builder
    .appName("kiosko-spark")
    .master("local[*]")
    .getOrCreate()
)
print(f"Spark version: {spark.version}\n")

print("Part 2 -- reading orders, dim_store, and dim_product with an explicit schema")
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(
    sorted(glob.glob("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)

print(f"orders_df.count() = {orders_df.count()}")
print(f"dim_store_df.count() = {dim_store_df.count()}")
print(f"dim_product_df.count() = {dim_product_df.count()}")
assert orders_df.count() == 40
assert dim_store_df.count() == 3
assert dim_product_df.count() == 4
print("Verification: 40 orders, 3 stores, 4 products -> OK\n")

Notice sorted(glob.glob(...)) instead of the wildcard pattern as a string — the same pattern from module 1's lesson 6 exercise 1, which avoids the benign FileNotFoundException warning in the log.

Part 3 — The join, verified with no row loss

print("Part 3 -- joining with the DataFrame API, with no rows lost or duplicated")
joined_df = orders_df.join(dim_store_df, "store_id").join(dim_product_df, "product_id")
print(f"joined_df.count() = {joined_df.count()}")
assert joined_df.count() == orders_df.count()
print("Verification: the join lost or duplicated not a single row -> OK\n")

Part 4 — revenue, with .withColumn()

print("Part 4 -- calculating revenue with .withColumn()")
fact_orders_df = (
    joined_df
    .withColumn("revenue", col("quantity") * col("unit_price"))
    .select("order_id", "store_id", "product_id", "quantity", "unit_price", "revenue", "order_ts")
)
fact_orders_df.orderBy("order_id").show(3, truncate=False)

Part 5 — Grouped by store and by product

print("Part 5 -- grouping by store and by product")
revenue_by_store = (
    fact_orders_df.groupBy("store_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
    .orderBy("store_id")
)
revenue_by_product = (
    fact_orders_df.groupBy("product_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"), F.count("*").alias("num_orders"))
    .orderBy("product_id")
)
revenue_by_store.show()
revenue_by_product.show()

Part 6 — Verification against the four previous guides

print("Part 6 -- verifying against dict/DuckDB/Polars/SQL (the four previous guides)")
EXPECTED_TOTAL = 106.15
EXPECTED_BY_STORE = {"S01": 38.3, "S02": 38.8, "S03": 29.05}
EXPECTED_BY_PRODUCT = {"P001": 33.55, "P002": 21.6, "P003": 10.5, "P004": 40.5}

spark_total = fact_orders_df.agg(F.round(F.sum("revenue"), 2).alias("total")).collect()[0]["total"]
spark_by_store = {r["store_id"]: r["total_revenue"] for r in revenue_by_store.collect()}
spark_by_product = {
    r["product_id"]: r["total_revenue"]
    for r in revenue_by_product.select("product_id", "total_revenue").collect()
}

assert spark_total == EXPECTED_TOTAL
assert spark_by_store == EXPECTED_BY_STORE
assert spark_by_product == EXPECTED_BY_PRODUCT
print(f"spark_total = {spark_total}")
print(f"spark_by_store = {spark_by_store}")
print(f"spark_by_product = {spark_by_product}")
print("Verification: all three levels match dict/DuckDB/Polars/SQL -> OK\n")

Parts 7 and 8 — Write the Parquet, reread and verify again

print("Part 7 -- writing fact_orders.parquet")
fact_orders_df.write.mode("overwrite").parquet("fact_orders.parquet")
print("Written: fact_orders.parquet\n")

print("Part 8 -- rereading the Parquet and verifying again")
reread_df = spark.read.parquet("fact_orders.parquet")
reread_count = reread_df.count()
reread_total = reread_df.agg(F.round(F.sum("revenue"), 2).alias("total")).collect()[0]["total"]
print(f"reread_df.count() = {reread_count}")
print(f"reread_total = {reread_total}")
assert reread_count == 40
assert reread_total == 106.15
print("Verification: fact_orders.parquet reread == 40 rows, total == 106.15 -> OK\n")

spark.stop()
print("=== spark.stop() -- module 3 closed, fact_orders.parquet ready for module 4 ===")

What to expect. Running python3 kiosko_fact_orders_in_spark.py in full (all eight parts together), the output is exactly this (executed in this run, PySpark 4.2.0):

=== Kiosko in Spark: fact_orders rebuilt, module 3 final delivery ===

Part 1 -- opening the SparkSession
Spark version: 4.2.0

Part 2 -- reading orders, dim_store, and dim_product with an explicit schema
orders_df.count() = 40
dim_store_df.count() = 3
dim_product_df.count() = 4
Verification: 40 orders, 3 stores, 4 products -> OK

Part 3 -- joining with the DataFrame API, with no rows lost or duplicated
joined_df.count() = 40
Verification: the join lost or duplicated not a single row -> OK

Part 4 -- calculating revenue with .withColumn()
+--------+--------+----------+--------+----------+------------------+-------------------+
|order_id|store_id|product_id|quantity|unit_price|revenue           |order_ts           |
+--------+--------+----------+--------+----------+------------------+-------------------+
|ORD-1001|S01     |P001      |3       |0.55      |1.6500000000000001|2026-08-03 08:14:00|
|ORD-1002|S01     |P002      |1       |1.2       |1.2               |2026-08-03 08:20:00|
|ORD-1003|S02     |P003      |2       |0.75      |1.5               |2026-08-03 08:31:00|
+--------+--------+----------+--------+----------+------------------+-------------------+
only showing top 3 rows
Part 5 -- grouping by store and by product
+--------+-------------+
|store_id|total_revenue|
+--------+-------------+
|     S01|         38.3|
|     S02|         38.8|
|     S03|        29.05|
+--------+-------------+

+----------+-------------+----------+
|product_id|total_revenue|num_orders|
+----------+-------------+----------+
|      P001|        33.55|        16|
|      P002|         21.6|        10|
|      P003|         10.5|         7|
|      P004|         40.5|         7|
+----------+-------------+----------+

Part 6 -- verifying against dict/DuckDB/Polars/SQL (the four previous guides)
spark_total = 106.15
spark_by_store = {'S01': 38.3, 'S02': 38.8, 'S03': 29.05}
spark_by_product = {'P001': 33.55, 'P002': 21.6, 'P003': 10.5, 'P004': 40.5}
Verification: all three levels match dict/DuckDB/Polars/SQL -> OK

Part 7 -- writing fact_orders.parquet
Written: fact_orders.parquet

Part 8 -- rereading the Parquet and verifying again
reread_df.count() = 40
reread_total = 106.15
Verification: fact_orders.parquet reread == 40 rows, total == 106.15 -> OK

=== spark.stop() -- module 3 closed, fact_orders.parquet ready for module 4 ===

Eight parts, eight checks, and the same 106.15 as always, now written to disk as fact_orders.parquet. Note the line only showing top 3 rows shows up glued to the next output with no blank line — that's Spark printing its own notice right before Part 5's print() runs; it isn't a formatting bug in this script.

Diagram: the eight parts, closing out the whole module

flowchart TD
    A["Part 1: SparkSession opened"] --> B
    B["Part 2 (L2): orders, dim_store,\ndim_product read -- 40/3/4"] --> C
    C["Part 3 (L3): chained join,\nno rows lost -- 40 == 40"] --> D
    D["Part 4 (L4): revenue calculated\nwith .withColumn()"] --> E
    E["Part 5 (L5): groupBy by store\nand by product, rounded"] --> F
    F["Part 6 (L6): verified against\ndict/DuckDB/Polars/SQL -- 3 asserts"] --> G
    G["Part 7-8 (L7): fact_orders.parquet\nwritten, reread, verified"] --> H["Module 3 closed:\nfact_orders.parquet ready\nfor module 4"]

Closing out this module's checklist, piece by piece

Six pieces resolved out of eleven — halfway through this guide, and the most important piece for your confidence: you know, with executed evidence, that Spark reproduces exactly the same numbers you already knew from three different engines.

Common mistakes

Turning in the project without Part 6's three checks. What happens: someone, in a hurry to reach Part 7 (writing the Parquet), runs Parts 1 through 5 and jumps straight to the write without verifying the result against the known values. Why it happens: Part 5's numbers on screen "look right," and formally verifying them seems like an extra step. How to spot it: if your final delivery doesn't include Part 6's three assert checks (total, by store, by product), you're writing a Parquet without having confirmed its content is correct — exactly the trap this module's lesson 6 warned about in detail. How to fix it: Part 6 isn't optional — it's the guarantee that makes everything after it trustworthy, including the Parquet file from Parts 7 and 8.

Writing the Parquet before verifying, instead of after. What happens: someone reorders the script's parts, writing fact_orders.parquet immediately after calculating revenue (Part 4), before grouping and verifying. Why it happens: it can seem more efficient to write the result as soon as it's ready, without waiting. How to spot it: if your fact_orders.parquet got written before any assert ran, you have no guarantee that file's content is correct — this project's part order (verify first, write after) isn't arbitrary. How to fix it: the correct order is always build, verify, and only then persist — never the other way around; an incorrect Parquet file, once other guides or pipelines start depending on it (like lakehouse-and-iceberg-guide in this same ecosystem), is far more expensive to fix than an in-memory DataFrame nobody else uses yet.

Copying the script without being able to explain why each part comes in that order. What happens: someone reuses kiosko_fact_orders_in_spark.py in their own work, changing only the file names, without being able to explain why Part 3 (join) comes before Part 4 (revenue), or why Part 6 (verification) comes before Part 7 (write). Why it happens: a script that already works is faster to copy than to understand from scratch. How to spot it: if you can't explain, without looking at the code, why this project verifies the result three times (total, by store, by product) instead of once, you need to reread lesson 6. How to fix it: every part of this project has a full lesson behind it that justifies it — before reusing this pattern in a project of your own, confirm you can explain each part in your own words.

Exercises

Exercise 1 — Extend the project with a Part 9: the breakdown by category. Without using datetime.now() (forbidden in this guide), add a ninth part to the script that groups by category and verifies the breakdown ({"beverages": 44.05, "electronics": 40.5, "snacks": 21.6}) — the fourth verification angle, over a column that comes from dim_product, not from fact_orders directly.

print("Part 9 -- breakdown by product category")
EXPECTED_BY_CATEGORY = {"beverages": 44.05, "electronics": 40.5, "snacks": 21.6}
spark_by_category = {
    r["category"]: r["total_revenue"]
    for r in (
        joined_df.withColumn("revenue", col("quantity") * col("unit_price"))
        .groupBy("category")
        .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
        .orderBy("category")
        .collect()
    )
}
print(f"spark_by_category = {spark_by_category}")
assert spark_by_category == EXPECTED_BY_CATEGORY
print("Verification: breakdown by category matches data-modeling -> OK")

Part 9 -- breakdown by product category
spark_by_category = {'beverages': 44.05, 'electronics': 40.5, 'snacks': 21.6}
Verification: breakdown by category matches data-modeling -> OK

Exercise 2 — Confirm fact_orders.parquet doesn't include store_name or product_name. Using spark.read.parquet("fact_orders.parquet").columns, confirm the written file only has the fact's seven columns, not the dimension attributes.

reread_columns = spark.read.parquet("fact_orders.parquet").columns
print(f"fact_orders.parquet columns: {reread_columns}")
expected_columns = ["order_id", "store_id", "product_id", "quantity", "unit_price", "revenue", "order_ts"]
assert reread_columns == expected_columns
print("Verification: fact_orders.parquet has exactly the fact's 7 columns -> OK")

fact_orders.parquet columns: ['order_id', 'store_id', 'product_id', 'quantity', 'unit_price', 'revenue', 'order_ts']
Verification: fact_orders.parquet has exactly the fact's 7 columns -> OK

Exercise 3 — Explain, without looking at the guide's design, what this pipeline is missing to become module 8's full capstone. In a 4-6 sentence paragraph, describe what transformations, checks, or decisions kiosko_fact_orders_in_spark.py is missing to become module 8's full distributed pipeline.

Summary and next step: the end of module 3

With this mini-project you close out module 3 in full. You read Kiosko's three tables with an explicit schema, joined them with .join() with no row lost or duplicated, calculated revenue with .withColumn() — seeing, again, the same floating-point phenomenon data-engineering-foundations-guide already documented — grouped by store and by product, and verified with three independent asserts that the result — 106.15 in total revenue, S01=38.3, S02=38.8, S03=29.05 — matches exactly what you already knew from dict, DuckDB, Polars, and SQL over a complete star schema. You closed by writing fact_orders.parquet to disk, this entire guide's first persistent artifact.

You took the most important step for trusting Spark: you didn't learn its syntax in the abstract — you used it to reproduce, with verified evidence, a result you already knew by heart. What you still haven't done is genuinely feel why distributing has a cost — forty rows fit, with no effort at all, in a single partition.

Where you're headed. Module 4 — partitions-and-the-cost-of-shuffle — deliberately and completely deterministically builds this guide's first dataset that actually needs to be distributed: kiosko_orders_at_scale, ten million rows, generated by replicating that same forty-order week once per synthetic "franchise." There you're going to see, for the first time with real evidence — not a promise — exactly what a shuffle is, and why it's the central cost of distributing anything.

Resources

• Apache Spark — SQL Getting Started (the full read-transform-verify pattern this project pulls together). spark.apache.org/docs/latest/sql-getting-started.html.
• Apache Spark — DataFrameWriter.parquet (the write reference used in Part 7 of this project). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameWriter.parquet.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — module 3's full objective and its place in the eight-module plan.
• python-for-data-engineering-guide DESIGN doc and data-modeling-for-analytics-guide DESIGN doc — the sources of the expected values (EXPECTED_TOTAL, EXPECTED_BY_STORE, EXPECTED_BY_PRODUCT) this project verifies with Spark. src/guides/python-for-data-engineering-guide/DISENO.md and src/guides/data-modeling-for-analytics-guide/DISENO.md