Module 3: Rebuilding Fact Orders With The Dataframe Api

Joining facts and dimensions with the DataFrame API

Next →

Description

This is the lesson where, for the first time in this guide, orders_df stops standing alone and crosses paths with dim_store_df and dim_product_df. You're going to use the DataFrame API's .join(), chained twice, to produce a single DataFrame with each order's complete information — and, with the same evidence discipline you already saw in data-modeling-for-analytics-guide, you're going to verify that join loses or duplicates not a single one of the forty rows. You're also going to see, with a real error captured on screen, a .join() behavior that surprises anyone coming from pure SQL.

Connection to the module. This lesson builds on the three tables read in lesson 2 — orders_df, dim_store_df, dim_product_df — and produces the piece lesson 4 needs to calculate revenue. Without this .join(), there's no store_name or product_name to show, and without the three columns together in a single DataFrame, lesson 5's .groupBy("store_id") would have nowhere to pull each store's readable name from.

An analogy: crossing three shelves to build a complete card

Think again of the previous lesson's archive, with its three separate shelves. An order card, on its own, says little: "ORD-1001, S01, P001, 3, 0.55, ..." — a pile of codes with no context. An archivist wanting to build a complete, readable card would have to walk to the second shelf, find store S01's card, copy its name and city; then walk to the third shelf, find product P001's card, copy its name and category; and finally staple the three cards into one complete one. That's exactly what .join() does for you, for all forty orders at once, without you having to walk to any shelf by hand: for every row in orders_df, it looks up the matching row in dim_store_df (by store_id) and in dim_product_df (by product_id), and staples them all into a single, wider row.

Worked example: the chained join

Step 1 — Pick back up the three tables from lesson 2

# join_facts_and_dims.py
from pyspark.sql import SparkSession
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)

Step 2 — The chained join

joined_df = (
    orders_df
    .join(dim_store_df, "store_id")
    .join(dim_product_df, "product_id")
)

print(f"orders_df.count() (before the join) = {orders_df.count()}")
print(f"joined_df.count() (after the two joins) = {joined_df.count()}")
assert orders_df.count() == joined_df.count(), "the join lost or duplicated rows"
print("Verification: the join lost or duplicated not a single row -> OK\n")

joined_df.printSchema()

print("\njoined_df, sorted by order_id, first 5 rows:")
joined_df.orderBy("order_id").show(5, truncate=False)

spark.stop()

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

orders_df.count() (before the join) = 40
joined_df.count() (after the two joins) = 40
Verification: the join lost or duplicated not a single row -> OK

root
 |-- product_id: string (nullable = true)
 |-- store_id: string (nullable = true)
 |-- order_id: string (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)

joined_df, sorted by order_id, first 5 rows:
+----------+--------+--------+--------+----------+-------------------+-------------+--------+---------------------+-----------+---------+
|product_id|store_id|order_id|quantity|unit_price|order_ts           |store_name   |city    |product_name         |category   |unit_cost|
+----------+--------+--------+--------+----------+-------------------+-------------+--------+---------------------+-----------+---------+
|P001      |S01     |ORD-1001|3       |0.55      |2026-08-03 08:14:00|Kiosko Centro|Bogota  |Bottled Water 600ml  |beverages  |0.4      |
|P002      |S01     |ORD-1002|1       |1.2       |2026-08-03 08:20:00|Kiosko Centro|Bogota  |Energy Bar           |snacks     |0.6      |
|P003      |S02     |ORD-1003|2       |0.75      |2026-08-03 08:31:00|Kiosko Norte |Lima    |Instant Coffee Sachet|beverages  |0.35     |
|P004      |S01     |ORD-1004|1       |4.5       |2026-08-03 09:02:00|Kiosko Centro|Bogota  |Phone Charger Cable  |electronics|2.1      |
|P001      |S03     |ORD-1005|5       |0.55      |2026-08-03 09:15:00|Kiosko Sur   |Santiago|Bottled Water 600ml  |beverages  |0.4      |
+----------+--------+--------+--------+----------+-------------------+-------------+--------+---------------------+-----------+---------+
only showing top 5 rows

Forty rows before, forty rows after two JOINs — the same "verify, don't assume" discipline you already saw in data-modeling-for-analytics-guide. And notice something in printSchema(): the leftmost column is product_id, not store_id or order_id. This isn't a bug — it's the documented behavior of .join()'s shorthand form: when you pass a column name as a string ("store_id", "product_id"), Spark merges that column into a single copy and moves it to the front of the resulting schema. The first .join(dim_store_df, "store_id") put store_id at the front; the second .join(dim_product_df, "product_id"), applied on top of that result, moves the most recent join's column — product_id — to the front again, pushing store_id to second position. The column order changed; the data didn't.

Diagram: two JOINs, one chained onto the other

flowchart LR
    A["orders_df\n40 rows"] -->|"join(dim_store_df,\n'store_id')"| B["orders + dim_store\n40 rows"]
    C["dim_store_df\n3 rows"] -->|"join(dim_store_df,\n'store_id')"| B
    B -->|"join(dim_product_df,\n'product_id')"| D["joined_df\n40 rows"]
    E["dim_product_df\n4 rows"] -->|"join(dim_product_df,\n'product_id')"| D

Going deeper: .join()'s shorthand form versus the explicit condition

Spark offers two ways of writing the same JOIN condition, and the difference between them isn't just style. The shorthand form, which this lesson uses, passes the column name as a string:

orders_df.join(dim_store_df, "store_id")

When both DataFrame objects have a column with exactly that name, Spark assumes it's the join column, and — this is what matters — merges the two columns into one in the result. That's why this lesson's joined_df has a single store_id column, not two.

The other form, the explicit equality condition, is what you'd use if the column names differed between the two tables (for example, if orders used store_id but dim_store used id_tienda):

orders_df.join(dim_store_df, orders_df.store_id == dim_store_df.store_id)

With this form, Spark does not merge anything — it keeps both columns, store_id from orders_df and store_id from dim_store_df, as two separate columns in the result, even though they share a name. This is exactly what produces the error in the next section.

Common mistakes

Using the explicit == condition when column names match, and running into an ambiguous reference. What happens: someone, used to always writing the full condition (df1.col == df2.col) out of SQL habit, uses it even when the column name is identical in both DataFrame objects — and when they later try to select that column by name, Spark can't decide which of the two copies they mean. Here it is, executed, exactly that error:

joined_explicit = orders_df.join(dim_store_df, orders_df.store_id == dim_store_df.store_id)
print("Columns:", joined_explicit.columns)
joined_explicit.select("store_id").show(1)

What to expect (executed in this run):

Columns: ['order_id', 'store_id', 'product_id', 'quantity', 'unit_price', 'order_ts', 'store_id', 'store_name', 'city']

Notice: store_id shows up twice in the column list. And when trying to select it:

pyspark.errors.exceptions.captured.AnalysisException: [AMBIGUOUS_REFERENCE] Reference `store_id` is ambiguous, could be: [`store_id`, `store_id`]. SQLSTATE: 42704

Why it happens: the == condition compares two columns' values, but doesn't tell Spark they're "the same conceptual column" — both survive, under the same name, in the result. How to spot it: if your .select() or your .groupBy() fails with AMBIGUOUS_REFERENCE right after a .join(), check how you wrote the JOIN condition. How to fix it: when the column names match between the two tables — Kiosko's case with store_id and product_id — always use the shorthand form (.join(dim_store_df, "store_id")), as this lesson's worked example does; save the explicit condition (==) for when the names genuinely differ between the two tables.

Assuming the column order in the result is still the same as in orders_df. What happens: someone writes code depending on a column's position (for example, joined_df.columns[0] expecting it to be order_id), and is surprised when the result doesn't match. Why it happens: in orders_df, order_id really was the first column — but, as you saw in the worked example, each .join() with the shorthand form moves the join column to the front. How to spot it: if your code breaks after adding a new .join(), and the error has to do with "the wrong column" at a specific position, suspect column order, not the data. How to fix it: never depend on a column's position — always use the name (joined_df["order_id"] or simply "order_id" in .select()), the exact same discipline you already learned with csv.DictReader in foundations, which avoids this problem entirely.

Joining against the wrong table first, with nothing failing. What happens: someone writes orders_df.join(dim_product_df, "store_id") by a copy-paste mistake — using the wrong join column for the table they're joining against. Why it happens: dim_product_df doesn't have a store_id column, so this particular code would in fact fail with a clear error (store_id doesn't exist) — but a subtler mistake, like joining dim_store_df using "product_id" by accident if that column happened to exist in both tables by coincidence, wouldn't be as obvious. How to spot it: always check that the join column you pass as a string makes semantic sense with the right-hand table — store_id with dim_store_df, product_id with dim_product_df — not just that the code runs with no error. How to fix it: the before/after row count (this lesson's worked example's assert) is the real safety net — if the join column were wrong but somehow didn't fail, an unexpected change in row count would be the first sign something's off.

Exercises

Exercise 1 — Confirm a LEFT JOIN produces the same result as the inner JOIN, for Kiosko's real data. Using .join(dim_store_df, "store_id", "left") instead of the default inner JOIN, confirm the row count is still 40 — evidence that every store_id in orders really does have a matching store in dim_store.

See solution 
left_joined_df = orders_df.join(dim_store_df, "store_id", "left")
print(f"left_joined_df.count() = {left_joined_df.count()}")
assert left_joined_df.count() == 40
print("Verification: LEFT JOIN gives the same count as INNER JOIN -> OK")

Expected output:

left_joined_df.count() = 40
Verification: LEFT JOIN gives the same count as INNER JOIN -> OK

If some store_id in orders had no match in dim_store, a LEFT JOIN would keep that row (with dim_store's columns as null) while an INNER JOIN would have dropped it — the fact that both give 40 confirms there's no orphaned store_id in Kiosko's real data.

Exercise 2 — Count how many rows the result would have if you joined on category instead of product_id. Without running anything yet, predict what would happen if you wrote orders_df.join(dim_product_df, "category") — note orders_df has no column called category. Then, verify your prediction by running the code.

See solution

Prediction: the code should fail, because "category" doesn't exist in orders_df — .join()'s shorthand form needs the column to exist on both sides.

try:
    bad_join = orders_df.join(dim_product_df, "category")
    bad_join.show(1)
except Exception as e:
    print(type(e).__name__, "-", str(e)[:200])

Expected output (executed in this run, abbreviated message):

AnalysisException - [UNRESOLVED_USING_COLUMN_FOR_JOIN] USING column `category` cannot be resolved on the left side of the join. The left-side columns: [`order_id`, `order_ts`, `product_id`, `quantity`, `store_id`, `unit_price`]. SQLSTATE: 42703

Confirmed: Spark can't resolve a join column that doesn't exist on the join's left side (orders_df has no category), and fails with a clear error — even listing the real available columns — instead of producing a silently empty or incorrect result.

Exercise 3 — Explain, without code, why the order of the two .join() calls (first dim_store, then dim_product) doesn't affect the final result, only the column order. In 2-3 sentences, justify why joining against dim_product_df first and dim_store_df second would produce exactly the same forty rows with the same values, even though printSchema() would look different.

See solution

The two JOINs are independent of each other: the first crosses orders against dim_store using store_id, and the second crosses that result against dim_product using product_id — neither depends on the other having run first, because they use different join columns that already existed in orders_df from the start. Changing the order only changes which join column ends up at the front of the resulting schema (as you saw in the worked example), but the final set of rows and values is identical either way — the same forty (store, product) pairs per order, regardless of the sequence they were resolved in.

Summary and next step

In this lesson you joined, for the first time, orders_df with dim_store_df and dim_product_df using .join() chained twice, verifying with an assert that the usual forty rows arrived with none lost or duplicated. You also saw, with a real error captured on screen, why .join()'s shorthand form (passing the column name as a string) is the right one when the names match between the two tables — and what happens, with AMBIGUOUS_REFERENCE, when you use an explicit equality condition instead with no need to.

Before moving on you should be able to: write this lesson's chained .join() from memory; explain the difference between .join()'s shorthand form and its explicit condition, and when to use each; and predict which column ends up at the front of the resulting schema after a chain of .join()s.

Lesson 4 takes this joined_df — forty rows, with all the store and product information already crossed in — and, finally, calculates revenue.

Resources

← PreviousReading `orders`, `dim_store`, and `dim_product` with an explicit schemaNext →Computing revenue, the same way, four engines later