Module 7: Parquet At Scale And Python Udfs

Module introduction: Parquet at scale and Python UDFs

Next →

Why this module exists

Modules 1 through 6 built Spark's complete in-memory mental model: the SparkSession, lazy evaluation, partitioning, shuffle, joins, window functions, and finally Catalyst and caching. But throughout that journey, fact_orders_at_scale never left memory as a partitioned artifact on disk — it got written just once, unpartitioned, in module 3 (fact_orders.parquet, forty rows). This module closes that debt twice over: first, it takes exactly the ten-million-row fact_orders_at_scale module 6 cached and queried three times, and writes it to disk partitioned by store_id — the way columnar Parquet actually gets used at scale, not as a flat file. Second, it resolves a question left pending since module 3: in dim_product, the unit_cost column got read, explained, and left there, with an explicit promise it "doesn't show up again until module 7, when you compute margin." This is that module.

There's a third piece, distinct from the previous two but related: up to now, every transformation in this guide got written with Spark's native DataFrame API — .select(), .filter(), .groupBy(), .join(), Window — functions Catalyst knows ahead of time and can optimize. This module introduces the first case where you need business logic that does not exist as a native Spark function: classifying every order as "high" or "low" margin, based on whether unit_price - unit_cost exceeds 1.0. You're going to solve that problem two ways — a plain Python UDF, and a pandas_udf vectorized with Arrow — and you're going to understand, with real execution-plan evidence (never a stopwatch), why the second way is the one that genuinely scales.

Connection to the module. This module doesn't revisit the in-memory partitions-and-shuffle mental model (module 4), nor joins/windows (module 5), nor Catalyst/caching (module 6) — it takes them for granted, and uses them as tools. What it adds is two new capabilities: writing and reading Parquet with a partitioning criterion, and extending the DataFrame API with your own Python logic, the right way.

Two analogies that are going to accompany this whole module

The first, for partitioned Parquet: picture a warehouse that, instead of piling all the boxes into a single room with no order, organizes them into three separate, labeled aisles from the moment they arrive — one aisle for S01, one for S02, one for S03. That work of organizing by aisle happens once, when storing (partitionBy("store_id") on the write). The payoff gets collected later, every time someone needs only S01's boxes: they walk straight into aisle S01, without even opening the door to the other two — that's partition pruning. And once inside the right aisle, if they only need each box's weight (not its complete contents), they can read the weight label without unpacking every one — that's predicate/column pushdown, lesson 3's topic.

The second, for UDFs: a plain Python UDF is like mailing a handwritten letter, different for each individual package, to a contractor overseas — the contractor has to open every envelope, read the letter, do the work, and write a reply back, once per package. A pandas_udf, by contrast, is a single shipping manifest: every package in a batch travels together, with one sheet describing all of them, and the contractor processes the complete batch in one pass, with no individual envelopes to open. The final work is the same — every package gets its classification — but the communication mechanism is radically different, and that difference is exactly what lessons 4 through 6 of this module teach you to read in Spark's execution plan.

Worked example: this module's map, before touching anything

# parquet_and_udfs_map.py
PARQUET_AND_UDFS_MAP = [
    (2, "Parquet at scale: partitioned writes",
        "fact_orders_at_scale (10M rows) written with partitionBy('store_id') -- 3 labeled aisles"),
    (3, "Predicate and column pushdown",
        "PartitionFilters (partition pruning) vs PushedFilters (real predicate) -- two distinct mechanisms, read in explain()"),
    (4, "Why plain Python UDFs are slow",
        "cloudpickle, row by row, BatchEvalPython in the plan -- the official quote on the bottleneck"),
    (5, "pandas_udf and Arrow-based vectorization",
        "pandas Series, processed in batches (block by block) -- ArrowEvalPython in the plan"),
    (6, "Rewriting margin_category as a pandas_udf",
        "The same logic, two mechanisms, one result -- compared side by side over the real week"),
    (7, "Naming Structured Streaming, without building it",
        "The same DataFrame API, applied to continuous data -- one paragraph, zero pipelines"),
    (8, "Project: partitioned Parquet and a vectorized UDF",
        "Kiosko's complete pipeline at scale, with margin_category computed over 10,000,000 rows"),
]

print("=== Parquet at scale and Python UDFs -- before touching anything ===\n")
for number, title, detail in PARQUET_AND_UDFS_MAP:
    print(f"L{number}. {title}")
    print(f"     {detail}\n")

print("This module's goal: write and read Parquet with a partitioning criterion,")
print("and know WHY a pandas_udf replaces a plain UDF at scale -- with evidence, never a stopwatch.")

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

=== Parquet at scale and Python UDFs -- before touching anything ===

L2. Parquet at scale: partitioned writes
     fact_orders_at_scale (10M rows) written with partitionBy('store_id') -- 3 labeled aisles

L3. Predicate and column pushdown
     PartitionFilters (partition pruning) vs PushedFilters (real predicate) -- two distinct mechanisms, read in explain()

L4. Why plain Python UDFs are slow
     cloudpickle, row by row, BatchEvalPython in the plan -- the official quote on the bottleneck

L5. pandas_udf and Arrow-based vectorization
     pandas Series, processed in batches (block by block) -- ArrowEvalPython in the plan

L6. Rewriting margin_category as a pandas_udf
     The same logic, two mechanisms, one result -- compared side by side over the real week

L7. Naming Structured Streaming, without building it
     The same DataFrame API, applied to continuous data -- one paragraph, zero pipelines

L8. Project: partitioned Parquet and a vectorized UDF
     Kiosko's complete pipeline at scale, with margin_category computed over 10,000,000 rows

This module's goal: write and read Parquet with a partitioning criterion,
and know WHY a pandas_udf replaces a plain UDF at scale -- with evidence, never a stopwatch.

Notice the structure: lessons 2 and 3 solve the Parquet-at-scale problem — how it gets written partitioned, and how that write decision pays off on the read afterward. Lessons 4 through 6 solve the UDF problem, in three steps: first the problem (why the naive path is slow, with plan evidence), then the solution (pandas_udf, with plan evidence), and finally the complete rewrite, side by side. Lesson 7 names, without building, the natural extension of everything you already know toward continuous data. Lesson 8 integrates the previous seven lessons into a single verified pipeline.

Diagram: where you were, where you're headed

flowchart LR
    subgraph M16["Modules 1-6 (already done)"]
        A["Spark installed, fact_orders rebuilt,\npartitions/shuffle measured,\njoins and windows at scale,\nCatalyst, explain(), caching with criteria"]
    end

    subgraph M7["This module (7 of 8)"]
        B["L2-L3: fact_orders_at_scale written\nwith partitionBy('store_id'),\nPartitionFilters vs PushedFilters in explain()"]
        C["L4-L6: why a plain @udf is slow\n(cloudpickle, row by row) --\n@pandas_udf vectorized with Arrow (Series, batched)"]
        D["L7: Structured Streaming named,\nnothing built"]
        E["L8: project -- partitioned Parquet\n+ pandas_udf over 10,000,000 rows"]
    end

    subgraph M8["Module 8"]
        F["Capstone: complete distributed pipeline,\ndecision tree -- does Kiosko need Spark?"]
    end

    A --> B --> C --> D --> E --> F

This module's map

Lesson    What it builds
────────  ──────────────────────────────────────────────────────────────
L1        (this one) The map: partitioned Parquet and UDFs, before touching them
L2        fact_orders_at_scale written with partitionBy("store_id")
L3        PartitionFilters (pruning) vs PushedFilters (real predicate), read in explain()
L4        Why a plain @udf is slow -- cloudpickle, BatchEvalPython
L5        @pandas_udf -- pandas Series, vectorized with Arrow, ArrowEvalPython
L6        margin_category rewritten as a pandas_udf, compared side by side
L7        Structured Streaming named -- the same API, continuous data
L8        Project: partitioned Parquet + pandas_udf over 10M rows

Lessons 2 and 3 answer the same question from two angles: how data gets organized when writing it (L2), and how that organization pays off when reading it (L3). Lessons 4, 5, and 6 aren't three separate topics — they're the same problem (margin_category) solved first the naive way, then the right way, and finally compared explicitly. Lesson 7 is deliberately the shortest in this entire guide: it names, without a single line of streaming code, where this same engine extends to. Lesson 8 integrates everything.

Going deeper: why Parquet and UDFs share a module

At first glance, partitioned Parquet and Python UDFs look like two unrelated topics — one is about how a file gets organized on disk, the other is about how Python code runs inside a DataFrame. But they share the same underlying question, the one that's run through this entire guide since module 1: what information does Spark have ahead of time, and what does it have to discover on the fly? A partitioned Parquet tells Spark, before reading a single byte, which physical folder every store_id value lives in — information a flat CSV can't offer. A plain Python UDF, by contrast, hides from Spark any information about what your function does: it's a black box Catalyst can't optimize, only execute, exactly as you already saw with RDDs in module 2. A pandas_udf doesn't change that underlying opacity — Spark still doesn't know what your Python logic does — but it radically changes the transport mechanism carrying the data to that black box: instead of one row at a time, a complete batch, with Arrow.

This module, then, doesn't break the discipline of the previous six — keep reading the execution plan as the only honest evidence — it extends it to two new questions, PartitionFilters/PushedFilters on one side, BatchEvalPython/ArrowEvalPython on the other.

Common mistakes

Assuming partitioning by store_id is always the right decision, for any future query. What happens: someone, after seeing partitionBy("store_id")'s benefit in this module, partitions any new table by the first column that comes to mind, without thinking about which filters are going to get used afterward. Why it happens: once you see the benefit with evidence (partition pruning), it's tempting to generalize "partitioning is good" with no condition attached. How to spot it: if your most frequent query filters by a column different from the one you used in partitionBy(...), partition pruning never kicks in — you're going to read all three complete folders anyway. How to fix it: this module's lesson 2 explains the complete criterion: store_id makes sense as a partition column in Kiosko because module 5 already established the real query pattern — "give me one store's data" — with the broadcast joins; partitioning by a column nobody filters afterward just multiplies the number of small files on disk, with no read benefit.

Confusing a Python UDF with a native Spark function, in terms of what Catalyst can optimize. What happens: someone expects .explain() over a DataFrame with an @udf to show the same level of detail — pushed predicates, columns pruned inside the function — that they already saw with native .filter() or .select(). Why it happens: a UDF gets written with the same .withColumn() syntax as any native transformation, so it's easy to assume Catalyst treats it the same way. How to spot it: if you look, inside a plan's BatchEvalPython/ArrowEvalPython node, for any predicate or optimized projection inside your Python function, you won't find one — Catalyst can't see inside your Python code, regardless of whether it's a plain UDF or a pandas_udf. How to fix it: a UDF — either type — is always an opaque box to Catalyst; the only thing that changes between a @udf and a @pandas_udf is how the data gets transported to that box and back, never whether Catalyst can optimize the logic inside. Lessons 4 through 6 of this module make that distinction explicit.

Expecting this module to teach you how to build a real streaming pipeline. What happens: someone reaches lesson 7, sees the name "Structured Streaming," and expects Kafka code, watermarks, or continuous triggers. Why it happens: the name sounds like a complete feature, and the rest of this guide does run real code for every topic it names. How to spot it: if you finish lesson 7 looking for a .py file with readStream to copy and run, check this guide's explicit boundary: real streaming is content for streaming-with-kafka-and-flink-guide, a complete sibling guide dedicated to that topic. How to fix it: this module's lesson 7 has a deliberately narrow goal — knowing it exists, and that the same DataFrame API you mastered in modules 1 through 7 extends to continuous data without changing vocabulary — not building a streaming system.

Exercises

Exercise 1 — Recite, from memory, this module's two analogies. Without rereading this lesson, explain in your own words the warehouse-with-labeled-aisles analogy (for partitioned Parquet) and the handwritten-letter-versus-shipping-manifest one (for UDFs). What decision gets made before the first query arrives, in each one?

See solution

The warehouse with labeled aisles represents partitioned Parquet: the decision to organize boxes by store gets made when storing (partitionBy("store_id") on the write), not when searching — by the time the first query filtering by store_id arrives, the separation work is already done, and all that's left is walking into the right aisle. The handwritten letter versus the shipping manifest represents the two UDF types: the decision of how to transport data to the Python code — row by row with cloudpickle, or in batches with Arrow — gets made when declaring the function, with @udf or with @pandas_udf, before it runs over a single real piece of data. In both cases, the early decision determines how expensive the repeated work that follows turns out to be.

Exercise 2 — Predict, before reading lesson 4, what concrete information Spark is missing when it runs a Python @udf, compared to .filter(col("quantity") > 1). Based on what you already learned about RDDs in module 2 (lesson 3) and about Catalyst in module 6, write your own prediction.

See solution

Reasonable prediction: when Spark runs .filter(col("quantity") > 1), the condition is a Column expression Catalyst can inspect, reorder, and even push down toward the data source (PushedFilters, as you already saw conceptually in module 3 and are going to see with real evidence in this module's lesson 3) — Spark "knows" what the filter does without running a single line of Python. An @udf, by contrast, wraps an arbitrary Python function: Spark only knows a function exists that takes certain columns and returns a value, but it can't look inside that function to optimize anything — the same structural limitation as module 2's RDDs, now applied to a single derived column instead of the whole DataFrame.

Exercise 3 — Explain, without code, why dim_product's unit_cost column has been present since module 3, but unused in any calculation until this module. In 2-3 sentences, using what you already know about Kiosko's star schema, explain what kind of business question needs unit_cost, and why that question hadn't shown up yet in this guide.

See solution

unit_cost (in dim_product) answers a different question from the one dominating modules 1 through 6: not "how much does Kiosko earn" (revenue = quantity * unit_price, the anchor figure for this entire guide), but "how profitable is each product" (margin = unit_price - unit_cost). Modules 3 through 6 focused on rebuilding and scaling the revenue calculation with different Spark tools — joins, aggregations, windows, caching — with no need yet for a derived column combining price and cost. This module needs it because margin_category is, deliberately, this guide's first calculation with no native Spark function ready to use — a simple condition (> 1.0) over a subtraction of two columns — and that's exactly why it's the perfect example for introducing UDFs: simple enough to verify by hand, new enough to justify a UDF instead of a native expression.

Summary and next step

This module closes two debts left pending from earlier modules: writing fact_orders_at_scale as partitioned Parquet (not flat, like module 3's fact_orders.parquet), and finally using the unit_cost column dim_product loaded since module 3 without touching it. Along the way, you're going to learn the exact mechanism — never measured with a stopwatch, always read in the execution plan — by which a vectorized pandas_udf replaces a plain Python UDF at scale.

Before moving on you should be able to: explain this module's two analogies in your own words (the warehouse with aisles, the letter versus the manifest); predict what information Catalyst is missing when it runs any UDF, regardless of type; and anticipate that lesson 7 names Structured Streaming without building anything, this guide's deliberate exception to the rest of its pattern.

Lesson 2 opens the first box: fact_orders_at_scale, the same ten million rows you cached in module 6, written for the first time in this guide with partitionBy("store_id").

Resources

← PreviousProject: Kiosko's optimized query planNext →Parquet at scale: partitioned writes