Module 1: Why Distribute The Single Node Ceiling

Reading Kiosko's orders with Spark

Next →

Description

This is the first time, in this entire guide, that Spark touches real Kiosko data. You're going to recreate the seven files orders_2026-08-03.csv through orders_2026-08-09.csv — identical, byte for byte, to the ones dict already processed in foundations, DuckDB and Polars in python-for-data-engineering, and SQL over a star schema in data-modeling — and read them with spark.read.csv(), declaring an explicit schema with StructType. You're also going to see, with executed evidence, two real Spark behaviors the official documentation recommends understanding before you trust them: what happens when you let Spark infer the schema instead of declaring it, and what happens — for real, not in theory — when a CSV's header doesn't match the schema you declared.

Connection to the module. This lesson uses the SparkSession you built in lesson 5 for its first real job. Lesson 7 verifies, with a second source of evidence, that this read's result is exactly correct.

An analogy: the public archive with a shared card format

Picture a public archive — a neighborhood library — with seven different folders, one for each day of the week, each full of loan cards. If each folder used a different card format, you'd have to read folder by folder, guessing what each column means in each one. But if the seven folders share exactly the same card format — the same six boxes, in the same order, with the same labels — you can treat them as a single collection: you pull all the cards out of the seven folders at once and process them together, without caring which folder each one came from. That's exactly what spark.read.csv("orders_2026-08-*.csv", ...) does with Kiosko's seven files: since they all share the same schema — order_id, store_id, product_id, quantity, unit_price, order_ts — Spark reads them as if they were a single file, without your code needing to know, or caring, which of the seven each row came from.

Worked example: reading the seven files with an explicit schema

Step 1 — Recreate Kiosko's seven files

Save these exact seven files, in the same folder where you're going to run your script — it's the same data, not a single value changed, that you already saw in foundations, python-for-data-engineering, and data-modeling:

# orders_2026-08-03.csv (8 orders, Monday)
order_id,store_id,product_id,quantity,unit_price,order_ts
ORD-1001,S01,P001,3,0.55,2026-08-03T08:14:00
ORD-1002,S01,P002,1,1.20,2026-08-03T08:20:00
ORD-1003,S02,P003,2,0.75,2026-08-03T08:31:00
ORD-1004,S01,P004,1,4.50,2026-08-03T09:02:00
ORD-1005,S03,P001,5,0.55,2026-08-03T09:15:00
ORD-1006,S02,P002,2,1.20,2026-08-03T09:47:00
ORD-1007,S03,P003,1,0.75,2026-08-03T10:05:00
ORD-1008,S01,P001,2,0.55,2026-08-03T10:22:00

The remaining six files (orders_2026-08-04.csv through orders_2026-08-09.csv) follow the same format, with 6, 2, 5, 7, 9, and 3 orders respectively — forty orders in total, the same fixed week you already know. If you'd rather generate them with code instead of copying them by hand, lesson 8 of this module includes the full script that rebuilds them.

Step 2 — Declare the explicit schema and read

# read_kiosko_orders.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),
])

orders_df = spark.read.csv(
    "orders_2026-08-*.csv",
    schema=orders_schema,
    header=True,
)

orders_df.printSchema()

print(f"orders_df.count() = {orders_df.count()}")

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

spark.stop()

What to expect. Running python3 read_kiosko_orders.py, the output is exactly this:

root
 |-- order_id: string (nullable = true)
 |-- store_id: string (nullable = true)
 |-- product_id: string (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- unit_price: double (nullable = true)
 |-- order_ts: timestamp (nullable = true)

orders_df.count() = 40

orders_df, sorted by order_id, first 5 rows:
+--------+--------+----------+--------+----------+-------------------+
|order_id|store_id|product_id|quantity|unit_price|order_ts           |
+--------+--------+----------+--------+----------+-------------------+
|ORD-1001|S01     |P001      |3       |0.55      |2026-08-03 08:14:00|
|ORD-1002|S01     |P002      |1       |1.2       |2026-08-03 08:20:00|
|ORD-1003|S02     |P003      |2       |0.75      |2026-08-03 08:31:00|
|ORD-1004|S01     |P004      |1       |4.5       |2026-08-03 09:02:00|
|ORD-1005|S03     |P001      |5       |0.55      |2026-08-03 09:15:00|
+--------+--------+----------+--------+----------+-------------------+
only showing top 5 rows

Forty rows, the exact same number as the three previous guides. The types from printSchema() confirm the explicit schema applied correctly: quantity as integer, unit_price as double, order_ts as a real timestamp — not text, which is what csv.DictReader handed back in foundations before manual conversion with dataclass. With spark.read.csv() and an explicit schema, that type conversion happens in the same read step, with no extra code.

Diagram: seven files, one shared schema, one DataFrame

flowchart LR
    A["orders_2026-08-03.csv\n8 rows"] --> G["Shared schema:\norder_id, store_id,\nproduct_id, quantity,\nunit_price, order_ts"]
    B["orders_2026-08-04.csv\n6 rows"] --> G
    C["orders_2026-08-05.csv\n2 rows"] --> G
    D["orders_2026-08-06.csv\n5 rows"] --> G
    E["orders_2026-08-07.csv\n7 rows"] --> G
    F["orders_2026-08-08.csv\n9 rows"] --> G
    H["orders_2026-08-09.csv\n3 rows"] --> G
    G --> I["orders_df\nA SINGLE DataFrame\n40 rows"]

Going deeper: explicit schema versus inferSchema=True

Spark can, if you ask it to, guess the schema instead of you declaring it:

df_inferred = spark.read.csv("orders_2026-08-*.csv", header=True, inferSchema=True)
df_inferred.printSchema()

What to expect (executed in this run):

root
 |-- order_id: string (nullable = true)
 |-- store_id: string (nullable = true)
 |-- product_id: string (nullable = true)
 |-- quantity: integer (nullable = true)
 |-- unit_price: double (nullable = true)
 |-- order_ts: timestamp (nullable = true)

On Kiosko's clean data, inferSchema=True gets all six types right, exactly the same as the explicit schema. This is worth saying honestly: the difference between the two ways of reading isn't, in this case, about correctness — both give the same result. The difference is about cost and control. To infer the schema, Spark has to read the data once extra, just to sample the values and guess their types, before the real read — a cost that grows with file size, and one an explicit schema avoids entirely, because Spark already knows, without guessing anything, what type to expect in each column. And there's a second, quieter difference: inference can get it right today, with today's data, and fail tomorrow if a new value doesn't fit the pattern Spark inferred (for example, if some day an order_id came in empty in a row, and that was the only row Spark used to infer the type). An explicit schema doesn't depend on which values show up first — it declares the intent up front, the exact same principle behind the dataclass conversion you already saw in foundations.

Common mistakes

Trusting that Spark reorders columns by name, the way csv.DictReader did. This is the most important trap in this lesson, and it's worth seeing with real evidence, not just as a warning. In foundations, you learned that csv.DictReader is safe even if someone reorders a file's columns, because it accesses each field by name. With spark.read.csv(schema=..., header=True), that isn't true by default. Try this: create a CSV with the first two columns swapped (store_id before order_id) and read it with this lesson's same schema:

# test file with swapped columns:
# store_id,order_id,product_id,quantity,unit_price,order_ts
# S01,ORD-9001,P001,1,0.55,2026-08-10T08:00:00

df_reordered = spark.read.csv("orders_reordered_test.csv", schema=orders_schema, header=True)
df_reordered.show(truncate=False)

What to expect (executed in this run):

+--------+--------+----------+--------+----------+-------------------+
|order_id|store_id|product_id|quantity|unit_price|order_ts           |
+--------+--------+----------+--------+----------+-------------------+
|S01     |ORD-9001|P001      |1       |0.55      |2026-08-10 08:00:00|
+--------+--------+----------+--------+----------+-------------------+

Look closely: the order_id column shows S01, and the store_id column shows ORD-9001 — swapped, exactly backward. Spark didn't throw any error — it only logged a warning (not an exception) saying CSV header does not conform to the schema... Expected: order_id but found: store_id, and kept reading the data by position, ignoring the file's real header. This is documented behavior, not a bug: Spark's CSV data source's enforceSchema option defaults to true, and Spark's official documentation says it plainly: "If it is set to true, the specified or inferred schema will be forcibly applied to datasource files, and headers in CSV files will be ignored" — and adds an explicit recommendation: "Though the default value is true, it is recommended to disable the enforceSchema option to avoid incorrect results."

How to fix it: pass enforceSchema=False explicitly. With that option, the same file with swapped columns produces a real error, instead of silently incorrect data:

df_strict = spark.read.csv("orders_reordered_test.csv", schema=orders_schema, header=True, enforceSchema=False)

What to expect (executed in this run, real Spark message, abbreviated):

org.apache.spark.SparkException: [FAILED_READ_FILE.NO_HINT] Encountered error while reading file ...
Caused by: org.apache.spark.SparkIllegalArgumentException: CSV header does not conform to the schema.
 Header: store_id, order_id, product_id, quantity, unit_price, order_ts
 Schema: order_id, store_id, product_id, quantity, unit_price, order_ts
Expected: order_id but found: store_id

An explicit error, with the exact name of the mismatched column, instead of misaligned data with no warning at all. This guide uses enforceSchema=False on every CSV read starting with this lesson, precisely for this reason — the same "fail loud, not silent" discipline you already saw in the previous guides in the ecosystem.

Interpreting the benign "FileNotFoundException" warning as a real error. What happens: when reading with a wildcard pattern ("orders_2026-08-*.csv"), Spark prints, before any result, a log block that includes the line java.io.FileNotFoundException: File orders_2026-08-*.csv does not exist, with a long trace beneath it. Someone seeing this for the first time assumes the read failed. Why it happens: internally, Spark first checks whether the given path corresponds to a structured-streaming metadata directory (a feature unrelated to this lesson) — that check tries to open the wildcard pattern as if it were a literal path, fails as expected, and Spark logs that expected failure as a WARN-level warning before continuing with normal pattern resolution. How to spot it: if your script finishes with the correct result (count() == 40, visible data with .show()) despite that block, the exception was just log noise, not a real failure — Spark caught it internally and moved on. How to fix it: nothing to fix in the code — but it's worth getting used to reading the final result before assuming any trace in the output is a fatal error; if you'd rather have cleaner output, pass an explicit list of files (spark.read.csv(sorted(glob.glob("orders_2026-08-*.csv")), ...)) instead of a wildcard string — with an explicit list, this particular warning doesn't show up.

Exercises

Exercise 1 — Read with an explicit list of files, not a wildcard pattern. Using glob.glob(), build the sorted list of Kiosko's seven files and pass it directly to spark.read.csv() instead of the string "orders_2026-08-*.csv". Confirm the count is still 40.

See solution 
import glob

files = sorted(glob.glob("orders_2026-08-*.csv"))
print("Files:", files)

orders_df = spark.read.csv(files, schema=orders_schema, header=True, enforceSchema=False)
print("count:", orders_df.count())

Expected output:

Files: ['orders_2026-08-03.csv', 'orders_2026-08-04.csv', 'orders_2026-08-05.csv', 'orders_2026-08-06.csv', 'orders_2026-08-07.csv', 'orders_2026-08-08.csv', 'orders_2026-08-09.csv']
count: 40

The result is identical to the wildcard pattern (40 rows), but without the benign FileNotFoundException warning from the common mistakes section — because Spark never tries to resolve a wildcard pattern when you've already given it the list of literal paths.

Exercise 2 — Confirm enforceSchema=False changes nothing when the data is fine. Read Kiosko's seven real files (with no column mismatch at all) with enforceSchema=False explicit, and confirm the result is identical to enforceSchema=True (the default).

See solution 
orders_df_strict = spark.read.csv("orders_2026-08-*.csv", schema=orders_schema, header=True, enforceSchema=False)
print("count with enforceSchema=False:", orders_df_strict.count())

Expected output:

count with enforceSchema=False: 40

When a file's header does match the declared schema — as with Kiosko's seven real files — enforceSchema=False changes absolutely nothing about the result; it only changes behavior when there's a real mismatch. That's why this guide always uses it: it costs nothing when the data is fine, and it prevents a silent error when it isn't.

Exercise 3 — Explain, without code, why printSchema() isn't enough to catch the swapped-columns problem. This lesson's common mistake showed that, with enforceSchema=True (the default), Spark reads a file with swapped columns without throwing any error. In 2-3 sentences, explain why checking only the output of orders_df.printSchema() wouldn't have been enough to catch this problem.

See solution

printSchema() shows the schema Spark applied — the column names and types you declared — it doesn't check whether those names actually correspond to each column's real data. In the common-mistake example, printSchema() would have shown order_id: string, store_id: string, exactly as declared, with no signal at all that the real values were swapped — because the schema was applied by position, not by content. The only way to catch the problem was to look at the values with .show() and notice that order_id held something shaped like a store_id ("S01") — the lesson here is that a correct schema doesn't guarantee the data is correctly aligned with it.

Summary and next step

In this lesson you read, for the first time with Spark, Kiosko's seven real files: spark.read.csv() with an explicit StructType, header=True, and — this module's most important common-mistakes lesson — enforceSchema=False, so a mismatch between the header and the schema fails with a clear error instead of silently mixing up columns. You confirmed orders_df.count() == 40, saw the correctly typed schema with printSchema(), and understood the real difference (cost, not correctness over clean data) between an explicit schema and inferSchema=True.

Before moving on you should be able to: write a StructType with orders's six fields from memory; explain what enforceSchema=False does and why this guide always uses it; and explain the difference between the benign FileNotFoundException warning with a wildcard pattern and a real Spark error.

Lesson 7 doesn't read anything new — it takes this lesson's orders_df and verifies it from a second angle, with pure Python, confirming that the same forty rows you see here really are the correct forty rows.

Resources

  • Apache Spark — SQL Getting Started (the general data-reading pattern with spark.read this lesson uses). spark.apache.org/docs/latest/sql-getting-started.html.
  • Apache Spark — CSV Files (specific documentation for the CSV data source: header, schema, enforceSchema with its default value and the explicit recommendation to disable it). spark.apache.org/docs/latest/sql-data-sources-csv.html.
  • data-engineering-foundations-guide DESIGN doc — the original source of Kiosko's seven files and their six-column schema. src/guides/data-engineering-foundations-guide/DISENO.md
← PreviousYour first SparkSessionNext →Verifying the same forty rows arrive