Project: Kiosko's first transformation chain

Description

This project closes the module by pulling the previous seven lessons together into a single script, verified end to end with assert: you open the SparkSession, read orders_df (lazy, no action), build a chain of four transformations, confirm with a real job count that nothing has executed yet, inspect the plan with .explain() (also executes nothing), trigger two different actions and confirm the job count climbs with each, and close with the historical RDD contrast that opened this module. The whole script runs for real, and every check is backed by an assert — not anyone's word for it.

Connection to the module. This project doesn't introduce any new concept — it's the final integration of lessons 2 through 7: who executes (driver/executors, lesson 2), which API (DataFrame over RDD, lessons 3-4), and when it executes (lazy evaluation, lessons 5-7), all verified with real evidence in a single flow.

An analogy: the whole day, from the list to the shopping bag

The previous lessons showed each piece separately: writing the list (transformations), reviewing it at no cost (.explain()), and finally paying at the register (actions). This project is the whole day, start to finish: you get to the store (open the SparkSession), pick up the catalog (orders_df, lazy read), build your full list while walking the aisles (the chain of four transformations), review it once more before the line (.explain()), and finally pay — twice, in fact, once to see the detailed receipt (.show()) and once just to confirm the total (.count()).

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

You need: the verified SparkSession from module 1 (appName="kiosko-spark", master("local[*]")), and the seven files orders_2026-08-03.csv through orders_2026-08-09.csv in the same folder where you're going to run the script (you already have them from module 1).

The reference solution, verified

Part 1 — Open the SparkSession and confirm the starting point

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

print("=== Kiosko in Spark: first transformation chain, module 2 delivery ===\n")

print("Part 1 -- opening the SparkSession")
spark = (
    SparkSession.builder
    .appName("kiosko-spark")
    .master("local[*]")
    .getOrCreate()
)
status = spark.sparkContext.statusTracker()
print(f"Jobs right after opening the SparkSession: {len(status.getJobIdsForGroup())}\n")

Part 2 — Read orders_df (lazy, still no action)

print("Part 2 -- reading orders_df (lazy read, no action yet)")
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),
])
orders_df = spark.read.csv(
    "orders_2026-08-*.csv", schema=orders_schema, header=True, enforceSchema=False,
)
print(f"Jobs after spark.read.csv(): {len(status.getJobIdsForGroup())}\n")

Notice spark.read.csv(...) itself isn't an action either — it only describes, in a plan, where the data is going to come from and with what schema, exactly as you already saw in module 1 (where .count() was the action that actually executed the read).

Part 3 — Build the four-transformation chain

print("Part 3 -- building the transformation chain (select, filter, filter, withColumn)")
busy_s01_df = (
    orders_df
    .select("order_id", "store_id", "product_id", "quantity", "unit_price")
    .filter(col("store_id") == "S01")
    .filter(col("quantity") > 1)
    .withColumn("is_bulk_order", col("quantity") >= 3)
)
print(f"type(busy_s01_df) = {type(busy_s01_df).__name__}")
print(f"Jobs after select+filter+filter+withColumn (FOUR transformations): {len(status.getJobIdsForGroup())}")
assert len(status.getJobIdsForGroup()) == 0, "a transformation should not trigger any job"
print("Verification: zero jobs triggered by transformations -> OK\n")

is_bulk_order is a new column, calculated with .withColumn() — it doesn't touch fact_orders or revenue (that starts in module 3); it's just one more, fourth transformation, to confirm the "zero jobs" pattern holds no matter how many transformations you chain.

Part 4 — Inspect the plan without executing anything

print("Part 4 -- inspecting the plan with .explain() (doesn't trigger a job either)")
busy_s01_df.explain()
print(f"Jobs after .explain(): {len(status.getJobIdsForGroup())}")
assert len(status.getJobIdsForGroup()) == 0, ".explain() should not trigger any job"
print("Verification: .explain() triggers no jobs -> OK\n")

Parts 5 and 6 — Trigger two different actions, verifying each adds work

print("Part 5 -- triggering the first action: .count()")
n = busy_s01_df.count()
jobs_after_count = len(status.getJobIdsForGroup())
print(f"busy_s01_df.count() = {n}")
print(f"Jobs after .count(): {jobs_after_count}")
assert jobs_after_count > 0, "an action SHOULD trigger at least one job"
assert n == 10
print("Verification: .count() == 10 and triggered at least one job -> OK\n")

print("Part 6 -- triggering a second action: .show()")
busy_s01_df.show()
jobs_after_show = len(status.getJobIdsForGroup())
print(f"Jobs after .show(): {jobs_after_show}")
assert jobs_after_show > jobs_after_count, "a second action should add new jobs"
print("Verification: .show() added new jobs on top of what .count() already had -> OK\n")

Part 7 — The RDD contrast, one last time

print("Part 7 -- RDD contrast, just to see it once more")
doubled = spark.sparkContext.parallelize([1, 2, 3]).map(lambda x: x * 2).collect()
print(f"parallelize([1,2,3]).map(lambda x: x*2).collect() = {doubled}")

spark.stop()
print("\n=== spark.stop() -- session closed ===")

What to expect. Running python3 kiosko_first_transformation_chain.py in full (all seven parts together), the output is exactly this (executed in this run, with no AssertionError at all):

=== Kiosko in Spark: first transformation chain, module 2 delivery ===

Part 1 -- opening the SparkSession
Jobs right after opening the SparkSession: 0

Part 2 -- reading orders_df (lazy read, no action yet)
Jobs after spark.read.csv(): 0

Part 3 -- building the transformation chain (select, filter, filter, withColumn)
type(busy_s01_df) = DataFrame
Jobs after select+filter+filter+withColumn (FOUR transformations): 0
Verification: zero jobs triggered by transformations -> OK

Part 4 -- inspecting the plan with .explain() (doesn't trigger a job either)
== Physical Plan ==
*(1) Project [order_id#0, store_id#1, product_id#2, quantity#3, unit_price#4, (quantity#3 >= 3) AS is_bulk_order#7]
+- *(1) Filter (((isnotnull(store_id#1) AND isnotnull(quantity#3)) AND (store_id#1 = S01)) AND (quantity#3 > 1))
   +- FileScan csv [order_id#0,store_id#1,product_id#2,quantity#3,unit_price#4] Batched: false, DataFilters: [isnotnull(store_id#1), isnotnull(quantity#3), (store_id#1 = S01), (quantity#3 > 1)], Format: CSV, Location: InMemoryFileIndex(7 paths)[...], PartitionFilters: [], PushedFilters: [IsNotNull(store_id), IsNotNull(quantity), EqualTo(store_id,S01), GreaterThan(quantity,1)], ReadSchema: struct<order_id:string,store_id:string,product_id:string,quantity:int,unit_price:double>


Jobs after .explain(): 0
Verification: .explain() triggers no jobs -> OK

Part 5 -- triggering the first action: .count()
busy_s01_df.count() = 10
Jobs after .count(): 2
Verification: .count() == 10 and triggered at least one job -> OK

Part 6 -- triggering a second action: .show()
+--------+--------+----------+--------+----------+-------------+
|order_id|store_id|product_id|quantity|unit_price|is_bulk_order|
+--------+--------+----------+--------+----------+-------------+
|ORD-6001|     S01|      P001|       6|      0.55|         true|
|ORD-6004|     S01|      P004|       2|       4.5|        false|
|ORD-1001|     S01|      P001|       3|      0.55|         true|
|ORD-1008|     S01|      P001|       2|      0.55|        false|
|ORD-5001|     S01|      P002|       2|       1.2|        false|
|ORD-5004|     S01|      P003|       2|      0.75|        false|
|ORD-5007|     S01|      P001|       2|      0.55|        false|
|ORD-2004|     S01|      P003|       3|      0.75|         true|
|ORD-4001|     S01|      P001|       4|      0.55|         true|
|ORD-7001|     S01|      P001|       2|      0.55|        false|
+--------+--------+----------+--------+----------+-------------+

Jobs after .show(): 5
Verification: .show() added new jobs on top of what .count() already had -> OK

Part 7 -- RDD contrast, just to see it once more
parallelize([1,2,3]).map(lambda x: x*2).collect() = [2, 4, 6]

=== spark.stop() -- session closed ===

Stop on Part 3, because that's the heart of this entire module, with an assert proving it: four chained transformations — select, filter, filter, withColumn — and the job count is still exactly 0. Part 4 confirms the same for .explain(). Only in Part 5, with the first real action, does the count finally climb — and the result, is_bulk_order, shows exactly what's expected: true for orders with quantity >= 3 (ORD-6001 with quantity=6, ORD-1001 with quantity=3, ORD-2004 with quantity=3, ORD-4001 with quantity=4), false for the rest.

Diagram: the seven parts, closing out the whole module

flowchart TD
    A["Part 1 (L2): SparkSession opened,\n0 jobs"] --> B
    B["Part 2 (L5): orders_df read,\nlazy read, 0 jobs"] --> C
    C["Part 3 (L3-L6): chain of FOUR\ntransformations, 0 jobs -- assert"] --> D
    D["Part 4 (L6): .explain(),\nfull plan, 0 jobs -- assert"] --> E
    E["Part 5 (L7): FIRST action,\n.count() == 10, jobs > 0 -- assert"] --> F
    F["Part 6 (L7): SECOND action,\n.show(), jobs climb again -- assert"] --> G
    G["Part 7 (L3-L4): RDD, one last time,\nfor contrast"] --> H["Module 2 closed:\ndriver/executors, DataFrame API,\nlazy evaluation -- ALL verified"]

Closing out this module's checklist, piece by piece

Common mistakes

Trusting Part 6 (.show()) to verify the business result, without checking the order. What happens: someone copies this project's .show() result and assumes ORD-6001 "is the first order from S01" in some chronological or business sense. Why it happens: seeing something first in a table feels like a signal of meaningful order. How to spot it: as you already saw in lesson 7, without an explicit .orderBy() in Part 3's chain, the order you see only reflects the internal partition-processing order — there's no .orderBy() anywhere in busy_s01_df. How to fix it: if you need a specific order for a real report, add it explicitly; this project deliberately leaves it unsorted, as a final reminder from lesson 7.

Turning in the project without the assert checks from Parts 3, 4, 5, and 6. What happens: someone, in a hurry to reach the final result, only runs the parts that produce visible output (.show() in Part 6) and skips the assert checks verifying the job count in the earlier parts. Why it happens: assert statements produce no visible output when they pass — they only fail loudly if something's wrong — so they seem dispensable. How to spot it: if your final delivery of this project doesn't include the four assert checks on the job count, you have no verifiable proof the lazy-evaluation model holds in your own code — only the hope that it does. How to fix it: this project's assert checks aren't decoration — they're the difference between "I think Spark is lazy" and "I verified, with my own code, that Spark is lazy." Keep them.

Reading Part 7 (RDD) as the start of a pattern that repeats for the rest of the guide. What happens: someone finishes this project, sees the final RDD contrast, and assumes the next modules are going to alternate between RDD and DataFrame API as convenient. Why it happens: Part 7 shows up at the end of the project, in a position that might suggest "this is what comes next." How to spot it: check module 3's explicit boundary in this guide's DESIGN doc — rebuilding fact_orders uses exclusively join(), groupBy(), and agg() from the DataFrame API. How to fix it: Part 7 of this project is, literally, the last time you're going to see sc.parallelize() in this guide — a closing historical contrast, not the start of a pattern. Module 3 begins with no mention of RDD at all.

Exercises

Exercise 1 — Add a Part 8 that confirms the is_bulk_order count. Without using datetime.now() (forbidden in this guide), add a part to the script that counts how many rows of busy_s01_df have is_bulk_order == True versus False, using .groupBy("is_bulk_order").count().

print("Part 8 -- count by is_bulk_order")
busy_s01_df.groupBy("is_bulk_order").count().show()

Part 8 -- count by is_bulk_order
+-------------+-----+
|is_bulk_order|count|
+-------------+-----+
|         true|    4|
|        false|    6|
+-------------+-----+

Exercise 2 — Repeat the full project with store_id == "S02" instead of "S01". Change Part 3's condition and confirm the new total count, checking it against the count by store you already know from the module 1 project (S02 had thirteen orders total).

busy_s02_df = (
    orders_df
    .select("order_id", "store_id", "product_id", "quantity", "unit_price")
    .filter(col("store_id") == "S02")
    .filter(col("quantity") > 1)
    .withColumn("is_bulk_order", col("quantity") >= 3)
)
n_s02 = busy_s02_df.count()
print(f"busy_s02_df.count() = {n_s02}")

busy_s02_df.count() = 11

Exercise 3 — Explain, from memory, what this pipeline is missing to become module 3's work. Without looking at the guide's DESIGN doc, describe in a 4-6 sentence paragraph what transformations or checks kiosko_first_transformation_chain.py is missing to become the full fact_orders rebuild you're going to build in module 3.

Summary and next step: the end of module 2

With this mini-project you close out module 2 in full. You verified, with real evidence and not intuition, the three pieces of Spark's execution model: the driver coordinates, the executors process (verified with the Spark UI's own REST API in lesson 2); the DataFrame API gives Catalyst information an RDD could never give it (lessons 3 and 4, with .explain()'s AttributeError on RDD as direct proof); and no transformation — no matter how many you chain — executes anything until an explicit action arrives, verified in this project with four assert checks on Spark's real job count.

You took the second step on an eight-module path: you already know how to read Kiosko data with Spark (module 1) and understand exactly how and when Spark executes work (module 2) — what you still don't have is a single line of code that calculates revenue with Spark, or that joins orders_df with dim_store or dim_product. That starts in module 3.

Where you're headed. Module 3 — rebuilding-fact-orders-with-the-dataframe-api — takes this module's mental model and applies it to the real work: reading orders, dim_store, and dim_product with an explicit schema, joining them with .join(), recalculating revenue = quantity * unit_price, aggregating with .groupBy().agg(), and verifying the result is exactly the same 106.15 you already saw with four different engines in the previous guides in the ecosystem.

Resources

• Apache Spark — Cluster Mode Overview (driver/executors, the foundation for lesson 2 and for this project). spark.apache.org/docs/latest/cluster-overview.html.
• Apache Spark — SQL Programming Guide (DataFrame vs RDD, the foundation for lessons 3 and 4). spark.apache.org/docs/latest/sql-programming-guide.html.
• Apache Spark — RDD Programming Guide (lazy evaluation, transformations vs actions, the foundation for lessons 5 through 7). spark.apache.org/docs/latest/rdd-programming-guide.html.
• Apache Spark — Monitoring and Instrumentation (the source of statusTracker(), used in every job-count assert in this project). spark.apache.org/docs/latest/monitoring.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — module 3's full objective, where this project's close points to.