Project: Kiosko's first distributed pipeline

Description

This is the complete spark-and-distributed-processing-guide's close. You have kiosko_orders_at_scale deterministically generated (M4), fact_orders_at_scale rebuilt with broadcast joins and window functions (M3, M5), Catalyst read through its phases with caching decided by real criteria (M6), partitioned Parquet with a vectorized pandas_udf (M7), and — from this very module — the complete pipeline assembled, verified with two independent paths, with every design decision justified, and with an honest verdict on whether Kiosko genuinely needed all of this. This final project has seven parts: it assembles the pipeline one last time, end to end, in a single SparkSession, and closes by applying the complete decision tree — the same close that opened this guide in its first module, now answered with the finished capstone in front of you.

Connection to the module. This project introduces no new concept — it's the final integration of this module's previous seven lessons, and of the seven modules preceding them. It explicitly picks back up the four deliverables from the brief that opened this module in lesson 2: the complete pipeline, proof of correctness, justified decisions, and the honest answer to the decision tree. By the time this project closes, all four have executed evidence, one last time, over the complete pipeline.

An analogy: the fleet's complete shift, audited start to finish

This module's lesson 1 compared this capstone to the first shift where the whole truck fleet works together. This project is that shift's exact close: the final confirmation, with every piece in place, that the complete fleet — the seven modules, built one by one — delivered the complete order, and the honest question, at the end of the shift, of whether a complete fleet was even needed for that specific order. There's no new piece to add in this project — only the final, executed confirmation that everything you built across this entire guide works together, and the final verdict on when to use it.

The material: everything this guide built, in a single project

You need kiosko_orders_at_scale.csv (generated in module 4 with generate_orders_at_scale(250_000)), dim_store.csv and dim_product.csv (module 3), and the seven real orders_2026-08-*.csv files (module 1), all in the same folder where you're going to run this script.

The verified reference solution

Part 1 and 2 — SparkSession and reading kiosko_orders_at_scale, dim_store, dim_product

# kiosko_first_distributed_pipeline.py
import glob
import os
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)
import pandas as pd

print("=== Final mini-project: Kiosko's first distributed pipeline ===\n")

print("Part 1 -- SparkSession")
spark = (
    SparkSession.builder
    .appName("kiosko-spark")
    .master("local[*]")
    .config("spark.driver.memory", "4g")
    .getOrCreate()
)
print(f"Spark version: {spark.version}\n")

print("Part 2 -- reading kiosko_orders_at_scale, dim_store, dim_product")
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)
raw_count = orders_at_scale_df.count()
print(f"orders_at_scale_df.count() = {raw_count}")
assert raw_count == 10_000_000

Part 3 — Broadcast join, revenue, caching justified by three queries

print("\nPart 3 -- broadcast join + revenue, caching (reuse = 3 queries)")
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(col("quantity") * col("unit_price"), 2))
)
fact_orders_at_scale_df.createOrReplaceTempView("fact_orders_at_scale")
fact_orders_at_scale_df.cache()

total = fact_orders_at_scale_df.agg(F.round(F.sum("revenue"), 2).alias("total")).collect()[0][0]
by_store = {
    r["store_id"]: r["total_revenue"]
    for r in fact_orders_at_scale_df.groupBy("store_id").agg(F.round(F.sum("revenue"), 2).alias("total_revenue")).collect()
}
store_window = Window.partitionBy("store_id").orderBy("order_ts", "franchise_id", "order_id")
running = fact_orders_at_scale_df.withColumn("running_total", F.round(F.sum("revenue").over(store_window), 2))
final_running = {
    r["store_id"]: r["max_running"]
    for r in running.groupBy("store_id").agg(F.max("running_total").alias("max_running")).collect()
}
print(f"total_revenue = {total}")
print(f"by_store = {by_store}")
print(f"final running_total per store = {final_running}")
assert total == 26_537_500.00
assert by_store == {"S01": 9_575_000.00, "S02": 9_700_000.00, "S03": 7_262_500.00}
assert final_running == {"S01": 9_575_000.0, "S02": 9_700_000.0, "S03": 7_262_500.0}
print("assert OK: total, per-store breakdown, and final running_total all three match\n")

Part 4 — Top-product-per-store-per-day ranking

print("Part 4 -- top-product-per-store-per-day ranking")
fact_with_day = fact_orders_at_scale_df.withColumn("order_day", F.to_date("order_ts"))
revenue_by_product_day = (
    fact_with_day.groupBy("store_id", "order_day", "product_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
)
rank_window = Window.partitionBy("store_id", "order_day").orderBy(F.desc("total_revenue"), "product_id")
top_products = (
    revenue_by_product_day.withColumn("rank", F.row_number().over(rank_window)).filter(col("rank") == 1)
)
num_top = top_products.count()
print(f"num_top (rows with rank == 1) = {num_top}")
assert num_top == 20
print("assert OK: 20 store x day combinations\n")

Part 5 — Partitioned Parquet and pandas_udf over the complete ten million rows

print("Part 5 -- Parquet partitioned by store_id + pandas_udf margin_category")
fact_orders_at_scale_df.write.mode("overwrite").partitionBy("store_id").parquet("fact_orders_at_scale_final.parquet")
fact_from_parquet = spark.read.parquet("fact_orders_at_scale_final.parquet")

s01_only = fact_from_parquet.filter(col("store_id") == "S01")
s01_count = s01_only.count()
assert s01_count == 4_000_000

@pandas_udf(StringType())
def margin_category(unit_price: pd.Series, unit_cost: pd.Series) -> pd.Series:
    return ((unit_price - unit_cost) > 1.0).map({True: "high", False: "low"})

classified_df = fact_from_parquet.withColumn("margin_category", margin_category(col("unit_price"), col("unit_cost")))
margin_counts = {
    r["margin_category"]: r["n"]
    for r in classified_df.groupBy("margin_category").count().withColumnRenamed("count", "n").collect()
}
print(f"s01_only.count() (partition pruning) = {s01_count}")
print(f"count by margin_category (pandas_udf) = {margin_counts}")
assert margin_counts == {"low": 8_250_000, "high": 1_750_000}
print("assert OK: 4,000,000 rows for S01, high=1,750,000, low=8,250,000\n")

fact_orders_at_scale_df.unpersist(blocking=True)
print("cache released with unpersist()\n")

Part 6 — The decision tree, closing the question that opened this guide

def should_distribute(dataset_size_gb: float, exhausted_single_node_tools: bool) -> dict:
    TB = 1000
    HUNDRED_TB = 100_000
    if dataset_size_gb < TB:
        verdict = "NO"
    elif dataset_size_gb >= HUNDRED_TB:
        verdict = "YES, PROBABLY"
    elif not exhausted_single_node_tools:
        verdict = "NOT YET"
    else:
        verdict = "IT DEPENDS -- MEASURE THE REAL COST"
    return {"dataset_size_gb": dataset_size_gb, "verdict": verdict}


print("Part 6 -- the decision tree, applied to real Kiosko and to Kiosko at scale")
real_bytes = sum(os.path.getsize(f) for f in sorted(glob.glob("orders_2026-08-*.csv")))
scale_bytes = os.path.getsize("kiosko_orders_at_scale.csv")
r_real = should_distribute(real_bytes / 1_000_000_000, False)
r_scale = should_distribute(scale_bytes / 1_000_000_000, False)
print(f"Real Kiosko ({real_bytes} bytes): {r_real['verdict']}")
print(f"kiosko_orders_at_scale ({scale_bytes:,} bytes): {r_scale['verdict']}")
assert r_real["verdict"] == "NO"
assert r_scale["verdict"] == "NO"
print("Neither real Kiosko nor this guide's synthetic dataset needed Spark by cost --")
print("Spark was used throughout this guide to LEARN the mechanism, not because volume demanded it.\n")

Part 7 — The final report

print("Part 7 -- final report")
print(f"Rows processed: {raw_count:,}")
print(f"Verified total revenue: {total:,}")
print(f"Breakdown by store: {by_store}")
print(f"Margin category: {margin_counts}")
print("\n=== spark.stop() -- spark-and-distributed-processing-guide closed ===")
spark.stop()

What to expect. Running python3 kiosko_first_distributed_pipeline.py, Parts 1 through 7 produce exactly this (executed in this run, PySpark 4.2.0, Java 17, end to end in a single SparkSession):

=== Final mini-project: Kiosko's first distributed pipeline ===

Part 1 -- SparkSession
Spark version: 4.2.0

Part 2 -- reading kiosko_orders_at_scale, dim_store, dim_product
orders_at_scale_df.count() = 10000000

Part 3 -- broadcast join + revenue, caching (reuse = 3 queries)
total_revenue = 26537500.0
by_store = {'S02': 9700000.0, 'S01': 9575000.0, 'S03': 7262500.0}
final running_total per store = {'S02': 9700000.0, 'S01': 9575000.0, 'S03': 7262500.0}
assert OK: total, per-store breakdown, and final running_total all three match

Part 4 -- top-product-per-store-per-day ranking
num_top (rows with rank == 1) = 20
assert OK: 20 store x day combinations

Part 5 -- Parquet partitioned by store_id + pandas_udf margin_category
s01_only.count() (partition pruning) = 4000000
count by margin_category (pandas_udf) = {'low': 8250000, 'high': 1750000}
assert OK: 4,000,000 rows for S01, high=1,750,000, low=8,250,000

cache released with unpersist()

Part 6 -- the decision tree, applied to real Kiosko and to Kiosko at scale
Real Kiosko (2236 bytes): NO
kiosko_orders_at_scale (601,305,672 bytes): NO
Neither real Kiosko nor this guide's synthetic dataset needed Spark by cost --
Spark was used throughout this guide to LEARN the mechanism, not because volume demanded it.

Part 7 -- final report
Rows processed: 10,000,000
Verified total revenue: 26,537,500.0
Breakdown by store: {'S02': 9700000.0, 'S01': 9575000.0, 'S03': 7262500.0}
Margin category: {'low': 8250000, 'high': 1750000}

=== spark.stop() -- spark-and-distributed-processing-guide closed ===

Seven parts, zero failed asserts, and the same number — 26,537,500.00 — you already saw in this module's lesson 3, in lesson 4, and in every mini-project from modules 3 through 7 before this one. Eight complete modules, and the number didn't change by a single cent. The only thing that changed, in every module, was how you got there: first with forty rows and a single .join(), then with ten million and a real shuffle, then with a broadcast criterion, with a justified cache, with partitioned Parquet, with a vectorized UDF — and, at the end, with the honest question of whether that whole path was needed.

Diagram: the complete guide, end to end

flowchart TD
    subgraph Herencia["Inherited from earlier guides"]
        A["106.15 -- foundations, python-for-data-engineering,\ndata-modeling: the same revenue, three engines"]
    end
    subgraph Guia["This guide: M1-M7, seven pieces"]
        B["M1: SparkSession + Java 17"]
        C["M2: lazy evaluation, DAG"]
        D["M3: DataFrame API -- 106.15, fourth engine"]
        E["M4: kiosko_orders_at_scale -- 10M rows, shuffle"]
        F["M5: broadcast join + windows"]
        G["M6: Catalyst + explain() + AQE + caching"]
        H["M7: partitioned Parquet + pandas_udf"]
    end
    subgraph Capstone["M8: this project"]
        I["Complete pipeline -- 26,537,500.00"]
        J["Correctness verified, 2 independent paths"]
        K["3 design decisions, with measured evidence"]
        L["Decision tree: NO (real), NO (synthetic)"]
    end
    A -->|"same logic, new engine"| B --> C --> D --> E --> F --> G --> H
    H --> I --> J --> K --> L

Closing out this entire guide's promise, point by point

And the promise from the market warning that opened this guide's DESIGN doc — "the list jumps straight from Python to Spark, which is the 2018 sequence... the most profitable skill isn't distributing: it's NOT distributing, and knowing how to justify it" — closed with the same kind of evidence every module demanded: not a claim, a command run and literal output.

Common mistakes

Considering the project done without running Part 6, trusting "it's already known" that Kiosko doesn't need Spark. What happens: someone, after seven modules and this module's own lesson 6, considers Part 6's result obvious and skips running it, treating the guide as closed at Part 5. Why it happens: the NO verdict already showed up, with the same criterion, in lesson 6 — repeating it here feels redundant. How to spot it: if your final delivery doesn't include running Part 6 within this specific script, you didn't confirm the decision tree stays consistent inside the complete, assembled pipeline, not just in an isolated experiment from lesson 6. How to fix it: this project's Part 6, just like earlier modules' mini-projects' final parts, exists to confirm every piece keeps working correctly when integrated with the others, not just in isolation.

Comparing only the total revenue, and not the per-store breakdown nor the margin_category count. What happens: someone confirms 26537500.0 == 26,537,500.00 between this project and earlier modules, and considers the delivery closed, without checking by_store or margin_counts. Why it happens: total revenue is the final report's most visible number. How to spot it: a pipeline with a subtle bug in the partition column or in the pandas_udf could, in theory, still give the same total by coincidence, while the per-store breakdown or margin classification end up wrong. How to fix it: this project's Part 7 includes all four pieces — count, total, per-store breakdown, margin_category — this entire guide's same "never trust a single aggregated number" discipline, repeated since module 3.

Closing the project unable to explain why Part 6's NO verdict doesn't contradict the rest of the capstone. What happens: someone runs the seven parts successfully, sees every number correct, including the NO verdict, and closes the guide unable to explain why building a complete distributed pipeline and then concluding it wasn't needed isn't a contradiction. Why it happens: a negative verdict at the end of a long project superficially feels like a defeat. How to spot it: if you can't explain, without looking at any lesson, why "knowing how to operate Spark" and "knowing when Spark is needed" are two complementary skills, not contradictory ones, you're missing this module's lesson 6's central point, not yet consolidated. How to fix it: every part of this project answers a different question — does the pipeline work? is it correct? is it well designed? was it needed? — and all four answers are true simultaneously: yes it works, yes it's correct, yes it's well designed, and no, Kiosko didn't need it by cost. None contradicts the other three.

Exercises

Exercise 1 — Simulate the complete report for a Kiosko with only two stores (remove S03 from the dataset), and confirm the rest of the pipeline still works. Filter orders_at_scale_df to exclude store_id == "S03" before Part 3, and confirm the new total and per-store breakdown are consistent with S03's absence.

orders_two_stores_df = orders_at_scale_df.filter(col("store_id") != "S03")
fact_two_stores_df = (
    orders_two_stores_df.join(dim_store_df, "store_id").join(dim_product_df, "product_id")
    .withColumn("revenue", F.round(col("quantity") * col("unit_price"), 2))
)
total_two_stores = fact_two_stores_df.agg(F.round(F.sum("revenue"), 2).alias("t")).collect()[0][0]
print(f"total without S03 = {total_two_stores}")
assert total_two_stores == round(9_575_000.00 + 9_700_000.00, 2) == 19_275_000.00
print("Verification: the total without S03 is exactly S01 + S02 -> OK")

total without S03 = 19275000.0
Verification: the total without S03 is exactly S01 + S02 -> OK

Exercise 2 — Calculate what percentage of total revenue the margin_category == "high" category represents. Using classified_df and total, calculate what fraction of the 26,537,500.00 corresponds to rows classified as "high" margin (only P004).

high_revenue = classified_df.filter(col("margin_category") == "high").agg(F.round(F.sum("revenue"), 2).alias("t")).collect()[0][0]
pct_high = round(high_revenue / total * 100, 2)
print(f"revenue for margin_category=high: {high_revenue}")
print(f"percentage of the total: {pct_high}%")

revenue for margin_category=high: 11250000.0
percentage of the total: 42.39%

Exercise 3 — Write, without looking at any earlier lesson, this entire guide's complete summary in a 6-8 sentence paragraph. Explain the complete path from module 1 to this final project, naming what each module resolved and how it connects to Part 6's final verdict.

Summary and next step: closing out this entire guide

With this mini-project you close out the complete spark-and-distributed-processing-guide. You assembled Kiosko's distributed pipeline in a single SparkSession: reading ten million rows, a broadcast join, caching justified by three real reuses, window functions for a running total and a ranking, Parquet partitioned by store_id, and margin_category computed with pandas_udf over the complete dataset. You confirmed, with assert, the same 26,537,500.00 you already knew, with the same exact per-store breakdown (S01=9,575,000.00/S02=9,700,000.00/S03=7,262,500.00) and margin-category breakdown (high=1,750,000/low=8,250,000). And you closed with the complete decision tree, applied with no favoritism even to this guide's own synthetic dataset: NO for real Kiosko, NO for kiosko_orders_at_scale.

Across eight modules, you took this guide's complete step toward distributed computing: from installing Spark and confirming it reads the same old week, to understanding its execution model, to rebuilding the same transformation with a fourth engine, to feeling a real shuffle's cost over ten million rows, to deciding with real criteria between JOIN strategies, to reading the Catalyst optimizer and caching only when it genuinely saves work, to writing partitioned Parquet and vectorizing business logic with Arrow, and — the final step, the one no 2018 "big data" guide taught — to knowing, with evidence and not intuition, when all that machinery is actually needed.

Where you're headed. Everything you built in this guide — the SparkSession, the DataFrame API, kiosko_orders_at_scale, joins and windows at scale, Catalyst and caching with real criteria, partitioned Parquet and pandas_udf, and the honest decision tree — exists today over the same Kiosko case that opened this story, five guides back, in data-engineering-foundations-guide. This module's lesson 7 already traced the complete map toward this ecosystem's sibling guides: real orchestration with Airflow, dbt-versioned transformation, the complete lakehouse with Iceberg, streaming with Kafka and Flink, organization-scale data governance, real cloud infrastructure, and a cluster's real cost. Each one takes this pipeline — assembled, verified, and with an honest criterion for when to use it — as its own starting point.

Resources

• PySpark — PyPI, the package page you installed in module 1, this entire guide's foundation. pypi.org/project/pyspark.
• Apache Spark — SQL Performance Tuning, the central reference for Catalyst, joins, and caching this final project integrates. spark.apache.org/docs/latest/sql-performance-tuning.html.
• PySpark — "Unleashing UDFs & UDTFs", the pandas_udf reference this project applies over the complete dataset. spark.apache.org/docs/latest/api/python/user_guide/udfandudtf.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this guide's complete eight-module map, including the market warning that motivated it from the start. src/guides/spark-and-distributed-processing-guide/DISENO.md.
• data-engineering-foundations-guide's DESIGN doc — the source for the Kiosko case and the original 106.15 this entire guide ended up verifying, a fourth engine later, over ten million rows. src/guides/data-engineering-foundations-guide/DISENO.md.