Module 3: Rebuilding Fact Orders With The Dataframe Api

Computing revenue, the same way, four engines later

Next →

Description

This is the lesson for the formula you've already written three times: revenue = quantity * unit_price. In foundations you wrote it with a for loop over dictionaries. In python-for-data-engineering you wrote it as a SQL expression inside DuckDB, and as a Polars expression. In data-modeling you wrote it inside a SELECT over the full star schema. Here you write it a fourth time, with the DataFrame API's .withColumn() in Spark — and you're going to see, with executed evidence, the same floating-point phenomenon you already documented in foundations, now reproduced with a distributed engine.

Connection to the module. This lesson takes lesson 3's joined_df — forty rows, with store_name, city, product_name, category, and unit_cost already crossed in — and adds the column that gives everything else business meaning: revenue. Without this column, lesson 5 would have nothing to sum with .groupBy().

An analogy: adding the subtotal to an already-assembled card

Pick back up the complete card you assembled in lesson 3 — stapled, with the store name and product name already copied in. It's missing one last piece: that line's subtotal, how much that order added up to. A cashier, with a hand calculator, would take the quantity and unit price off the card, multiply them, and jot the result in the corner. .withColumn("revenue", col("quantity") * col("unit_price")) does exactly that, for all forty cards at once: it takes two columns that already exist in each row, multiplies them, and adds the result as a new column — without touching any of the columns that were already there.

Worked example: .withColumn() and computing revenue

Step 1 — Pick back up lesson 3's join

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

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

Step 2 — .withColumn() with the usual formula

fact_orders_df = joined_df.withColumn("revenue", col("quantity") * col("unit_price"))

print("fact_orders_df.printSchema():")
fact_orders_df.printSchema()

print("\nFirst 5 rows, sorted by order_id:")
fact_orders_df.select(
    "order_id", "store_id", "product_id", "quantity", "unit_price", "revenue"
).orderBy("order_id").show(5, truncate=False)

spark.stop()

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

fact_orders_df.printSchema():
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)
 |-- revenue: double (nullable = true)

First 5 rows, sorted by order_id:
+--------+--------+----------+--------+----------+------------------+
|order_id|store_id|product_id|quantity|unit_price|revenue           |
+--------+--------+----------+--------+----------+------------------+
|ORD-1001|S01     |P001      |3       |0.55      |1.6500000000000001|
|ORD-1002|S01     |P002      |1       |1.2       |1.2               |
|ORD-1003|S02     |P003      |2       |0.75      |1.5               |
|ORD-1004|S01     |P004      |1       |4.5       |4.5               |
|ORD-1005|S03     |P001      |5       |0.55      |2.75              |
+--------+--------+----------+--------+----------+------------------+

Look closely at the first row: revenue is 1.6500000000000001, not 1.65. This is not a bug in your code, or in Spark — it's the same floating-point phenomenon you already documented in data-engineering-foundations-guide, with the exact same multiplication (3 * 0.55), in a different language. revenue ended up typed as double, the same 64-bit binary representation (IEEE 754) Python uses, and 3 * 0.55 has no exact binary representation in that format — regardless of whether the multiplication is done by the Python interpreter, DuckDB's engine, or the JVM behind Spark. It's binary arithmetic, not a bug in any of the four tools.

Diagram: a new column, calculated from ones that already existed

flowchart LR
    A["joined_df\nquantity, unit_price,\n... (already-joined columns)"] -->|".withColumn('revenue',\ncol('quantity') * col('unit_price'))"| B["fact_orders_df\nsame columns + revenue"]

Going deeper: why this guide doesn't "fix" floating point in this lesson

You could, at this very step, wrap the formula in F.round(col("quantity") * col("unit_price"), 2) so each individual row shows 1.65 instead of 1.6500000000000001. This lesson, on purpose, doesn't do that yet — for the same reason data-engineering-foundations-guide didn't either, in its own derived-columns lesson. Rounding each individual row hides the phenomenon instead of teaching it, and it also raises a real design question: is the "correct" value to store in fact_orders the exact result of the binary multiplication, or an already-rounded-to-two-decimals version? The answer this guide — and the three previous ones in the ecosystem — chose is: store the exact, unrounded value, and round only when displaying or aggregating, never when calculating. Lesson 5 applies exactly that criterion: the sum gets calculated over the exact values, with no row-by-row rounding, and rounding gets applied exactly once, at the end, on the aggregated total.

This matters because rounding too early can accumulate error: if you rounded each of the forty rows before summing them, the total could differ by a few cents from the total you get by summing the exact values first and rounding afterward — a real problem in financial systems, not an abstract concern. Lesson 6 of this module verifies, with evidence, that summing first and rounding after gives exactly 106.15, the same number as the three previous guides.

Common mistakes

Assuming 1.6500000000000001 means the calculation is wrong. What happens: someone sees that number in the output and, without the context of the ecosystem's previous guides, assumes there's a bug in their .withColumn(), or that Spark calculated something different from what was expected. Why it happens: a sixteen-digit number where two were expected looks, at first glance, like an error. How to spot it: if the extra number shows up consistently on the same operation (3 * 0.55, or any multiplication of two double values whose exact result isn't representable in binary), and does not show up on other rows with different multiplications (like 1 * 1.20 = 1.2, exact), it's the documented floating-point phenomenon, not a bug. How to fix it: nothing to fix in the calculation itself — lesson 5 shows how to round correctly, when aggregating, not when calculating each row.

Calculating revenue with unit_cost instead of unit_price, after lesson 3's JOIN. What happens: since joined_df now has both columns — unit_price (from orders) and unit_cost (from dim_product) — someone, by mistake or by not checking the full name, writes col("quantity") * col("unit_cost") instead of col("quantity") * col("unit_price"). Why it happens: the two names are similar, and before lesson 3's JOIN this mix-up wasn't even possible — unit_cost didn't exist yet in orders_df. How to spot it: if your total revenue, calculated in lesson 5 or 6, comes out much lower than 106.15 (for example, close to the cost instead of the price), suspect this column first. How to fix it: revenue always uses unit_price; unit_cost doesn't show up again in any calculation until module 7, when you calculate margin (unit_price - unit_cost).

Forgetting .withColumn() doesn't modify joined_df, it returns a new DataFrame. What happens: someone writes joined_df.withColumn("revenue", ...) without assigning the result to a new variable, and is later surprised that joined_df doesn't have the revenue column. Why it happens: in some libraries (or in Python code with explicit mutation), it's common for a method to modify the original object. How to spot it: if after calling .withColumn() you still see the old schema when running joined_df.printSchema(), check whether you assigned the result to a new variable (fact_orders_df = joined_df.withColumn(...), as in the worked example) or just called the method without saving the result. How to fix it: like almost all of Spark's DataFrame API, .withColumn() is immutable — it always returns a new DataFrame, never modifies the original; always store the result in a variable.

Exercises

Exercise 1 — Calculate revenue by hand for two orders and compare. Without running any code yet, calculate by hand the revenue for ORD-1004 (quantity=1, unit_price=4.50) and for ORD-2006 (quantity=6, unit_price=0.55). Then, verify with .filter() on fact_orders_df.

See solution

By hand: ORD-1004 → 1 * 4.50 = 4.50. ORD-2006 → 6 * 0.55 = 3.30.

fact_orders_df.filter(
    col("order_id").isin("ORD-1004", "ORD-2006")
).select("order_id", "quantity", "unit_price", "revenue").show()

Expected output:

+--------+--------+----------+------------------+
|order_id|quantity|unit_price|           revenue|
+--------+--------+----------+------------------+
|ORD-1004|       1|       4.5|               4.5|
|ORD-2006|       6|      0.55|3.3000000000000003|
+--------+--------+----------+------------------+

ORD-1004 gives exactly 4.5 (this particular multiplication produces no floating-point artifact), while ORD-2006 does show 3.3000000000000003 instead of 3.30 — the same phenomenon from the lesson, on a different row.

Exercise 2 — Add a second derived column: revenue_per_unit. Using .withColumn() again, add a revenue_per_unit column that's simply revenue / quantity (which, mathematically, should give back unit_price). Confirm whether the result always matches unit_price exactly, or whether floating point affects this division too.

See solution 
from pyspark.sql.functions import col

with_ratio_df = fact_orders_df.withColumn("revenue_per_unit", col("revenue") / col("quantity"))
with_ratio_df.filter(col("order_id") == "ORD-1001").select(
    "order_id", "quantity", "unit_price", "revenue", "revenue_per_unit"
).show(truncate=False)

Expected output:

+--------+--------+----------+------------------+-----------------+
|order_id|quantity|unit_price|revenue           |revenue_per_unit |
+--------+--------+----------+------------------+-----------------+
|ORD-1001|3       |0.55      |1.6500000000000001|0.55             |
+--------+--------+----------+------------------+-----------------+

Interesting: even though revenue does show the artifact (1.6500000000000001), dividing that same value by quantity (3) gives back exactly 0.55 — the division "undoes" the binary rounding in this particular case. This isn't a general guarantee (not every floating-point operation is this convenient), but it confirms the underlying revenue value is still, for every practical purpose, mathematically correct.

Exercise 3 — Explain, without code, why this guide rounds when aggregating and not when calculating each row. In 2-3 sentences, explain the risk of rounding revenue to two decimals immediately after .withColumn(), instead of waiting to round the aggregated result in lesson 5.

See solution

If you round each of the forty rows individually before summing them, each row's rounding error (however small, typically less than a cent) can accumulate differently than if you sum the exact values first and round once, at the end. In practice, with numbers as small as Kiosko's, the difference would probably be invisible — but in a real financial system, with millions of rows, that accumulated difference can be material. That's why this guide, following the same criterion as the three previous guides in the ecosystem, stores revenue unrounded and applies ROUND/.round() only when displaying or aggregating, never when calculating each individual row.

Summary and next step

In this lesson you calculated revenue = quantity * unit_price with .withColumn(), the exact same formula you already wrote with dict, SQL, and Polars in the three previous guides. You saw, with executed evidence, the same floating-point phenomenon (1.6500000000000001 for ORD-1001) you already documented in foundations — confirming it isn't a Spark-specific behavior, but binary arithmetic in any engine — and you understood why this guide stores the exact, unrounded value, leaving rounding for when it aggregates.

Before moving on you should be able to: write the .withColumn() line that calculates revenue from memory; explain why 1.6500000000000001 isn't a bug; and justify why this guide rounds when aggregating, not when calculating each row.

Lesson 5 takes fact_orders_df — with revenue already calculated for the forty rows — and groups it, for the first time, by store and by product.

Resources

← PreviousJoining facts and dimensions with the DataFrame APINext →Grouping by store and by product