9. Narrow vs Wide Transformations

Narrow vs Wide Transformations

A Spark transformation is narrow when each output partition depends on data from exactly one input partition — no data needs to move between executors, so Spark can compute it without a shuffle. This is the cheaper, more parallel-friendly class of transformation, as opposed to a wide transformation (groupBy, join, distinct, repartition, …), where an output partition depends on data from multiple input partitions and a shuffle is required.

This isn’t a new idea, even coming from Ab Initio. To get a correct Rollup, you first repartition the data with PBK (Partition By Key), laid out all-to-all, so every record sharing the same key lands on the same partition before Rollup runs on it. That’s the same problem Spark is solving with a shuffle before a groupBy — redistribute so same-key rows land together, then aggregate locally. The mechanism looks different (Ab Initio wires it as an explicit graph component, Spark does it implicitly via hashpartitioning), but the underlying need — same key, same partition — is identical.

Here’s what that looks like as an actual graph — a two-phase Rollup around the repartition:

graph LR
    subgraph Local["Before repartition — local"]
        A["Input File (mfile)"] --> B["Filter By Expression"]
        B --> C["Reformat"]
        C --> D["Rollup<br/>key: customer_id<br/>Local Aggregation"]
    end
    D --> E["Partition By Key"]
    E -->|"data redistribution (all-to-all)"| F
    subgraph Shuffled["After repartition"]
        F["Rollup<br/>key: customer_id<br/>Final Aggregation"] --> G["Output File (mfile)"]
    end

Everything up to the local Rollup runs independently within each partition — no data movement yet, just shrinking the volume by pre-aggregating whatever happens to already be on that partition. Partition By Key is where records actually move — every row is redistributed so all rows sharing a customer_id land on the same partition. Only then can the final Rollup produce a correct, complete aggregate per key. It’s the same shape as Spark’s shuffle: a cheap partial aggregation first, a genuine data-movement step, then a final aggregation on correctly-grouped data.

Equivalent PySpark Physical Plan (approx.)

The same job in PySpark — df.filter(...).groupBy("customer_id").agg(F.sum("amount")) — compiles to a plan shaped exactly like the graph above:

graph LR
    subgraph Stage0["Stage 0"]
        A["FileScan parquet"] --> B["Filter"]
        B --> C["Project"]
        C --> D["HashAggregate<br/>partial_sum(amount)"]
    end
    D -->|shuffle files| E["Exchange<br/>hashpartitioning(customer_id, 200)"]
    subgraph Stage1["Stage 1"]
        E --> F["HashAggregate<br/>finalmerge_sum(amount)"]
        F --> G["Write Output"]
    end

• Stage 0 (everything before the Exchange): FileScan → Filter → Project → partial HashAggregate, all task-local, one task per input partition — no shuffle yet. This is Filter By Expression → Reformat → local Rollup.
• Exchange hashpartitioning(customer_id, 200) is the shuffle boundary, and it’s doing exactly what Partition By Key does: redistributing rows all-to-all so every row with the same customer_id lands on the same downstream partition. Spark just triggers it implicitly off groupBy instead of it being a wired component.
• Stage 1 (everything after the Exchange): the final HashAggregate merges the partial aggregates per key into the true, complete aggregate — the final Rollup — then writes the output.

An Exchange — like Partition By Key — is always a phase boundary: everything upstream has to finish before anything downstream can start reading its share. Spark can pipeline operators within a Stage but never across a shuffle.