How Distributed Aggregations Combine Results Across Workers

In F02, we learned how to distribute records across partitions so that multiple workers can process data in parallel. Now let’s consider a different requirement: calculating the total weekly revenue for each customer. Our orders dataset contains 21,154 events, including 4,400 orders placed by 797 customers. The revenue report needs to produce one result per customer, showing the total order amount in dollars for the week.

For example, if a customer placed three orders worth 500, 700, and 300 dollars, the expected total would be: 500 + 700 + 300 = 1,500 dollars

This calculation is straightforward when all three orders are available to one process. But what happens when the orders are distributed across multiple workers?
In our case, our dataset is divided into 12 partitions using event_id. Each partition is processed independently by a task. Because we’re partitioning by event ID rather than customer ID, events belonging to the same customer may be distributed across multiple partitions.
For example, suppose a customer has three orders:

Each task sees only the records in its assigned partition. Therefore:

• Task 0 calculates 500 dollars.
• Task 3 calculates 700 dollars.
• Task 8 calculates 300 dollars.

All three results are correct for their individual partitions, but none represents the customer’s complete weekly total. If we publish these partial results directly, the revenue report will be incorrect.

How to Calculate the Correct Total?

We need to combine the results from tasks that processed records belonging to the same customer. This requires three steps:

1. 01Map stage: Each task reads its partition and calculates partial revenue totals for the customers it encounters.
2. 02Shuffle: The system redistributes those partial totals so all results belonging to the same customer reach the same destination task.
3. 03Reduce stage: The destination task combines the partial totals to calculate the customer’s complete weekly revenue.

For our example: 500 + 700 + 300 = 1,500 dollars
The final result now includes all three orders, even though different tasks processed them.

In distributed processing, individual tasks often see only a portion of the data. A task may calculate a perfectly correct result for its partition, but that doesn’t mean the result is correct for the entire dataset. Some aggregations, such as SUM, COUNT, MIN, and MAX, can combine partial results directly. Others require additional information to produce the correct final answer. For example, calculating an average requires both the sum and count, while counting distinct values requires tracking unique values across workers rather than simply adding their individual counts.

We’ll explore these differences throughout this lesson.

In the previous section, we learned why calculating revenue across multiple partitions requires combining results from different tasks. Let’s now understand how a distributed system performs this calculation using two processing stages: Map and Reduce, connected by a shuffle.

Our nightly revenue job processes:

• 4,400 orders placed by 797 customers.
• 12 input partitions, each processed by a separate map task.
• 12 workers, allowing the map tasks to run in parallel.

Our goal is to calculate one complete weekly revenue total for each customer.

Calculating Weekly Revenue Total Per Customer

Step 1: Calculate Partial Revenue – Map Stage

Each map task reads its assigned partition and processes only records where event_type = 'order_placed'. Since a customer’s orders may be spread across multiple partitions, each task can calculate only the revenue from the orders it sees. Also, to avoid producing one intermediate row for every order, the task uses a combiner. A combiner performs local aggregation, maintaining a running sum for each customer within the task. For example, suppose Task 0 processes these records:

Instead of producing four separate records, the combiner produces three partial results:

The combiner reduces the number of intermediate records before data needs to move between tasks.

What Happens in Our Dataset?
In the reference simulator:

• Task 0 processes 403 orders and produces 226 partial customer totals.
• Across all 12 map tasks, the combiners produce 2,628 partial results.
• These partial results represent 797 unique customers.

Why do we have 2,628 results when there are only 797 customers? Because the same customer can appear in multiple map tasks. For example:

• cust-0001 has orders in all 12 tasks.
• Another 204 customers have orders in only one task.

Every map task produces one partial total for each customer it encounters. Therefore, 2,628 partial results do not mean 2,628 customers. They represent different portions of revenue for the same 797 customers.

Step 2: Bring Related Results Together - Shuffle

After the map stage, we need to combine all partial results belonging to the same customer. This is where the shuffle comes in. The shuffle redistributes intermediate results so that all partial totals for a customer reach the same reduce task. Our simulator uses the partitioning rule introduced in F02:
Reduce partition = stable_hash(customer_id) % 12

For example, all partial results belonging to cust-0001 are routed to Reduce Task 7. It doesn’t matter which map task originally calculated the partial revenue. The shuffle ensures that all partials for the same customer arrive at the same destination.

Step 3: Calculate the Final Revenue – Reduce Stage

Once the partial results have been brought together, the reduce task combines them into a single total. Consider cust-0001, who placed 263 orders across the 12 input partitions. Some of the partial revenue totals are:

The reduce task adds all 12 partial totals to produce: Total weekly revenue for cust-0001 = 5,547,205 dollars
After the reduce stage, the job produces exactly 797 results, one for each customer.

Our nightly revenue job contains:

• 4,400 orders
• 797 customers
• 12 map tasks
• A combiner that maintains one partial revenue total per customer in each task

How Does the Shuffle Affect Job Runtime?

So far, we’ve focused on correctness. But moving data between tasks also adds processing overhead. Our job now has two stages:

1. 01Map stage: Read records and calculate partial revenue.
2. 02Reduce stage: Receive shuffled partial results and merge them into final totals.

In our simulator, the reduce stage begins after all map tasks have completed. This creates a stage barrier, where the next stage must wait for the previous stage to finish. Our reference model produces the following runtime breakdown:

Therefore: Total runtime = 10 + 40 + 36 + 110.8 + 2.9 = 199.7 seconds

The shuffle and reduce introduce additional processing and data movement, but they are necessary to calculate the correct final totals for this input distribution. In our simulator, the shuffle also incurs approximately 0.02 units of network transfer cost. This is a simplified example, real distributed engines may overlap some shuffle data transfers with map execution.

Now let’s examine an incorrect result that can silently corrupt a distributed aggregation. The growth team notices that the revenue job spends additional time shuffling and merging partial results. During testing, they observe that removing the reduce stage doesn’t change the output for their sample dataset. Believing the reduce stage is unnecessary, they remove it to improve performance.
The modified job now:

1. 01Reads orders across 12 map tasks.
2. 02Calculates partial customer revenue using combiners.
3. 03Writes those partial results directly as the output.

The shuffle and reduce stage no longer run. The job finishes faster, and no execution error occurs. However, the revenue dashboard starts reporting incorrect totals.

What Went Wrong?

Remember that our map stage produces 2,628 partial results for 797 customers. Some customers appear in multiple map tasks because their orders are spread across different partitions. Without the reduce stage, these partial totals are never combined. To make matters worse, the dashboard loader expects one row per customer. When it encounters multiple rows for the same customer, it keeps only the first row and ignores the rest. As a result, the dashboard displays only a portion of each affected customer’s revenue.

Consider cust-0001. This customer has a true weekly revenue of 5,547,205 dollars, spread across 12 map tasks.
Some partial totals are:

• Task 0: 397,275 dollars
• Task 1: 409,549 dollars
• Task 2: 574,629 dollars

Understanding the Revenue Error

Let’s calculate how much of cust-0001’s revenue is actually displayed.
Displayed revenue percentage = (397,275 ÷ 5,547,205) × 100 ≈ 7.2%

The dashboard reports only 7.2% of the customer’s actual weekly revenue. The remaining 92.8% is present in other map tasks, but those partial results are discarded by the dashboard loader. This problem affects every customer whose orders appear in multiple partitions.
In our dataset:

• 797 customers have orders.
• 593 customers have orders distributed across multiple map tasks.
• Those customers can have incomplete totals when the dashboard retains only one partial result.

How Much Revenue Is Missing Overall?

The correct weekly revenue is: 75,387,030 dollars
However, without the reduce stage, the dashboard reports only: 17,110,437 dollars
That is approximately: (17,110,437 ÷ 75,387,030) × 100 ≈ 22.7%

The dashboard is reporting only 22.7% of the actual weekly revenue. This is a serious correctness problem, even though all map tasks completed successfully.

Why Didn’t the Job Fail?

From the execution engine’s perspective, nothing went wrong.
Each map task:

• Read its input records.
• Calculated valid partial revenue totals.
• Successfully wrote its output.

The problem is that partial results were treated as final results. The processing engine cannot automatically determine that the business requires a complete weekly total for each customer. That correctness requirement must be reflected in the computation and verified through tests.

The Job Actually Became Faster

Compare the two executions:

• With shuffle and reduce: 199.7 seconds.
• Without shuffle and reduce: 196.8 seconds.

Removing the reduce stage saves approximately 2.9 seconds. But the faster job produces incorrect revenue totals. This makes the issue particularly dangerous: performance improves, the job reports success, and the dashboard still receives data. The error may remain unnoticed until someone reconciles the dashboard with the original orders.

What Should We Learn from This?

When multiple tasks calculate partial values for the same key, those values must be combined before publishing the final aggregate. We should validate not only that the job runs successfully, but also that:

• Each customer has the expected final result.
• The aggregation output contains one row per customer.
• Customer revenue matches an independently calculated reference result.
• Changing the input partitioning does not change the final revenue totals.

We’ll explore why the team’s original test failed to detect this problem in the next section.

In the previous section, we discovered that removing the reduce stage caused the revenue dashboard to report incorrect totals. But there’s an important question:
Why did the team’s tests pass if the calculation was incorrect?

The answer lies in how the test data was partitioned.

How the Test Data Hid the Problem

During testing, the team partitioned the orders dataset using customer_id instead of event_id. As we learned in F02, partitioning by customer_id ensures that all events belonging to the same customer are assigned to one partition. This changes how our aggregation behaves.
With event_id partitioning:

• A customer’s orders may be spread across multiple map tasks.
• Each task calculates only part of that customer’s revenue.
• A reduce stage is needed to combine those partial results.

With customer_id partitioning:

• All orders belonging to a customer are processed by the same task.
• That task can calculate the customer’s complete weekly revenue.
• No additional merge is needed for that customer.

This means the pipeline without a reduce stage can produce correct results when the input is already partitioned by customer_id. However, it fails when the same customer’s records are distributed across multiple tasks.

Why Is the Customer-Partitioned Job Slower?

Notice another important difference. The map stage takes:

• 110.8 seconds with event_id partitioning.
• 222.7 seconds with customer_id partitioning.

This happens because customer-based partitioning introduces the data skew we studied in F02.2. Some customers generate significantly more events than others, causing certain partitions to contain much more work. In our dataset, the skew factor is 2.15 when hashing by customer_id into 12 partitions. As a result, one task takes much longer than the others, delaying the entire processing stage.

Compare two correct approaches:

• Partition by customer_id without a reduce stage: 308.7 seconds.
• Partition by event_id with a shuffle and reduce stage: 199.7 seconds.

For this workload, distributing the input more evenly and then combining customer totals is approximately 109 seconds faster.

How Should We Test Distributed Aggregations?

The team’s original test failed because its input distribution made the missing reduce stage unnecessary. To catch this type of problem, we need two kinds of tests.

1. Correctness Test: Compare Against a Trusted Result
Calculate the expected customer revenue using a single-process implementation. Then compare the distributed output with that result, customer by customer.
For every customer: Distributed revenue = Reference revenue

This verifies that the distributed computation produces the expected result.

2. Partitioning Test: Change How Records Are Distributed
Test the same aggregation using different input distributions. For example:

• Partition by customer_id, keeping customer records together.
• Partition by event_id, spreading customer records across tasks.
• Use a fixture where at least one customer’s orders appear in several partitions.

The final revenue totals should remain the same regardless of how the input is partitioned. If the results change when we change partitioning, the computation may be incorrectly relying on records being colocated.

The problem was that the pipeline produces partial customer revenue totals but doesn’t combine them into final results. The solution is to restore the reduce stage, which merges partial results belonging to the same customer. But how should we combine these results? That depends on the type of calculation.

Understanding Mergeable States

A state is the information a task maintains while processing records. For example, when calculating total revenue, each map task maintains a running sum for every customer it encounters. Suppose two tasks produce these partial results for cust-0001:

• Task 0: 397,275 dollars.
• Task 1: 409,549 dollars.

We can combine these values: 397,275 + 409,549 = 806,824 dollars

This produces the combined revenue from both tasks. We can continue merging the remaining partial totals until we obtain the customer’s complete weekly revenue of 5,547,205 dollars. This works because addition allows us to combine partial sums without needing the original individual records. We call this a mergeable state.

Which Calculations Can Be Merged?

Different aggregations require different intermediate states.

Note: The intermediate state is not always the same as the final result. For example, an average is a single number, but calculating it correctly across partitions requires keeping both a sum and a count. Similarly, a distinct count is a single number, but adding distinct counts from different partitions can produce incorrect results. Let’s explore that problem.

Why Can’t We Add Distinct Counts?

Suppose we want to calculate the number of unique customers who placed an order on March 2, 2026. The correct answer in our dataset is 325 distinct customers. Each of our 12 map tasks processes a different partition.
The team proposes the following approach:

1. 01Each map task counts the distinct customers in its partition.
2. 02The reduce stage adds those 12 counts.
3. 03The result becomes the total number of distinct customers.

This seems reasonable, but there’s a problem. The same customer may appear in multiple partitions. For example, a customer who placed orders appearing in five different tasks could be counted once by each of those five tasks. When we add the partial counts, that customer contributes five instead of one.

The Correct Approach: Merge Sets, Not Counts

Instead of maintaining only the number of distinct customers, each map task should maintain a set of customer IDs. A set stores unique values.
For example:
Task 0: {cust-001, cust-002, cust-003}
Task 1: {cust-002, cust-003, cust-004}

If we count the values separately:

• Task 0: 3 customers.
• Task 1: 3 customers.
• Sum of counts: 6 customers.

But the actual number of unique customers is 4. Instead, combine the sets:
{cust-001, cust-002, cust-003, cust-004}

Now count the unique values. Final distinct count = 4

The same approach works for our full dataset. Each task produces a set of customer IDs, and the reduce stage combines those sets before calculating the final count.

This gives the correct result: 325 distinct customers

The important distinction is:

• Incorrect: Count each set, then add the counts.
• Correct: Combine the sets, then count the unique values.

What Does Merging Sets Cost?

Keeping complete sets of customer IDs requires more memory and data movement than keeping only a numeric count. Our reference simulator illustrates this tradeoff at the scaled job size:

Moving sets requires substantially more data because the system must transfer customer identifiers rather than only numeric counts. In the reference model, the reduce work for merging these sets takes approximately 1.3 seconds. The extra cost is necessary for this exact set-based approach because the reduce stage must know which customer IDs appeared in multiple partitions. More advanced systems may use compact data structures to estimate distinct counts using less memory, but those estimates involve different accuracy and resource tradeoffs.

Why Does the Choice of State Matter?

A distributed aggregation needs an intermediate representation that can be safely combined with other partial results. Consider these two examples.

1. Average order value: Don’t average the averages produced by individual tasks. Instead, maintain: (sum_of_amounts, number_of_orders)
Merge both values across tasks, then calculate: Final average = Total sum ÷ Total count

2. Distinct customer count: Don’t add the distinct counts produced by individual tasks. Instead, maintain: set(customer_ids)
Merge the sets, then calculate: Final distinct count = Number of unique IDs in the merged set

In both cases, the final result is calculated after the intermediate states have been merged.

Not every final aggregation result can be combined directly across partitions.

To produce correct distributed aggregations, maintain a state that can be merged safely:

• Sums and counts can be added.
• Minimums and maximums can be compared.
• Averages require both sum and count.
• Exact distinct counts require preserving enough information to identify repeated values across partitions, such as sets of IDs.

The general principle is simple: Merge the intermediate states first, then calculate the final result.

Our experiments demonstrate three important lessons about distributed aggregations.

1. 01Map tasks produce partial results: When records belonging to the same customer are distributed across multiple tasks, each task calculates only part of that customer’s total. These partial results must be combined to produce the correct final result.
2. 02Skipping the merge can silently produce incorrect results: Our revenue job finished faster without the reduce stage, but the dashboard displayed incomplete customer totals. A successful execution doesn’t guarantee correct results.
3. 03Not every aggregation result can be merged directly: Sums and counts can be added, but averages and distinct counts require additional information. The intermediate state must contain enough information to calculate the correct result after merging.

Limitations of the Model

Our simulator makes several simplifying assumptions:

• Proportional workload scaling: The model scales record counts and customer counts proportionally when moving from the small fixture to the larger nightly workload. For example, its 797 customers represent approximately 2.26 million customers at the 60-million-event scale. Real datasets may grow differently.
• Fixed data distribution: The model assumes the relative distribution of customer activity remains unchanged. In production, new high-volume customers or changes in ordering patterns can affect partition sizes and processing costs.
• Simplified shuffle execution: Reduced tasks read the intermediate results without additional sorting or disk spilling. Real distributed engines may incur extra memory, sorting, and disk I/O costs.
• No failures or retries: Every task completes successfully. We’ll explore how distributed systems recover from failures and safely repeat computations in F04.
• Exact aggregation methods: We use sets to calculate exact distinct counts. Approximate algorithms can use smaller, mergeable representations to estimate distinct counts with less memory.

We’ve learned that some intermediate values can be combined directly, while others require additional information. Let’s examine two more aggregation examples.

Example 1: Why Averaging Partial Averages Can Be Wrong

Suppose we want to calculate the average order value for March 2, 2026. Our dataset contains 627 orders for that day, with a correct average order value of 17,630.4 dollars. The orders are distributed across 12 map tasks.

A developer proposes this approach:

1. 01Each map task calculates the average order value for its partition.
2. 02The reduce stage collects the 12 averages.
3. 03It calculates the average of those 12 values.

This seems reasonable, but there’s a problem. Not every task processes the same number of orders. When we average the task-level averages, each task contributes equally, regardless of how many orders it processed.

The Correct Approach: Merge Sum and Count

Instead of maintaining only the average, each map task should maintain two values:

• Sum: The total order amount processed by the task.
• Count: The number of orders processed by the task.

For example: (total_amount, order_count)
The reduce stage combines these pairs by adding their sums and counts separately. Across all 12 tasks, our dataset contains:

• Total order amount: 11,054,273 dollars
• Total orders: 627

Therefore: Global average = Total order amount ÷ Total order count

11,054,273 ÷ 627 ≈ 17,630.4 dollars

This produces the correct result. The important distinction is:

• Incorrect: Calculate an average in each task, then average those results without considering their counts.
• Correct: Merge the partial sums and counts, then divide once at the end.

The same problem can be difficult to notice when the difference is small. For example, across the entire week, the incorrect average-of-averages is 17,143.0 dollars, compared with the correct 17,133.4 dollars. The numbers are close, but the calculation is still wrong.

Example 2: Finding the Latest Event for Each Customer

Now suppose we want to find the most recent event timestamp for every customer. Each map task examines the customer events it processes and keeps the latest event_time_utc. Because a customer’s records may appear in multiple tasks, we still need to combine their partial results. Fortunately, maximum values can be merged directly.

For example:

• Task 0’s latest event: 10:30 AM
• Task 1’s latest event: 11:15 AM
• Task 2’s latest event: 9:45 AM

The reduce stage simply chooses the latest timestamp among these values. Final latest event time = 11:15 AM

This works because the maximum of the partial maximum values is also the maximum of the complete dataset. The same principle applies to minimum values.

How can we recognise when a distributed aggregation is producing incomplete or incorrect results? In our workbench, look for four warning signs:

1. 01Incorrect result comparison: The distributed output doesn’t match a trusted single-process calculation.
2. 02More output rows than expected: An aggregation intended to produce one result per customer returns multiple rows for the same customer.
3. 03Unexpected performance improvements: A job suddenly becomes faster after a processing stage is removed or modified, but no one verifies that the results remain correct.
4. 04Results change with partitioning: Running the same aggregation with a different input partitioning strategy produces different final totals.

What Does This Look Like in Production?

Similar problems may appear as:

• Revenue totals that disagree with source records or financial reports.
• Duplicate customer IDs in an output expected to contain one row per customer.
• Distinct customer counts increasing when the number of processing tasks changes.
• Average values change when records are distributed differently.
• Tests pass on customer-partitioned data but fail when customer records are spread across workers.

For example, our buggy revenue pipeline produced 2,628 partial results for 797 customers. Without the reduce stage, the dashboard displayed only 22.7% of the correct total revenue because it retained only one partial result per customer. The job still completed successfully, making the error easy to overlook.

What Should You Investigate?

When an aggregation produces unexpected results, examine:

• Input distribution: Can records belonging to the same key appear in multiple tasks?
• Intermediate results: Does each map task produce a partial result or a complete result?
• Merge logic: Are partial states combined correctly before publishing the output?
• Output correctness: Does the final result match a trusted reference calculation?
• Test coverage: Does the test dataset include keys that appear across multiple partitions?

Suppose our reporting team wants to identify the latest event timestamp for each customer. The input data is distributed across 12 map tasks. Each task calculates the maximum event_time_utc for every customer it encounters. Can the reduce stage combine these partial maximum values to find the correct latest timestamp?

Try It Yourself

When the F03 workbench is available, run:

The reference implementation should report: AGREE

This means:

• The output contains 797 customer keys.
• Every customer’s revenue matches the trusted single-process calculation.

Now run the version without the reduce stage:

The expected result is: DISAGREE

The buggy implementation produces 2,628 partial rows, and cust-0001 is the first customer whose result differs from the reference calculation. Compare the two executions and examine why the second version produces incorrect results.

Note: The F03 workbench is planned. These commands are intended for use once it becomes available.

Distributed aggregations often require combining results calculated by multiple tasks. When records belonging to the same key are spread across partitions, individual tasks can calculate only partial results. In this lesson, we learned that:

1. 01Map tasks calculate partial results: Our 12 map tasks produced 2,628 partial customer totals for 797 customers.
2. 02Shuffles bring related results together: Partial results belonging to the same customer are routed to the same reduce task.
3. 03Reduce tasks calculate final results: The partial states are merged to produce one complete result per customer.
4. 04Skipping the merge can silently corrupt results: Our buggy pipeline ran faster but caused the dashboard to display incomplete revenue totals.
5. 05Intermediate states must be mergeable: Sums, counts, minimums, maximums, and appropriately structured states can be combined correctly. Averages and distinct counts require more information than their final numeric values alone.

Our Nightly Revenue Example

The correct two-stage job:

• Processes 4,400 orders from 797 customers.
• Produces 2,628 partial results during the map stage.
• Merges them into 797 final customer totals.
• Completes in 199.7 seconds.

When the reduce stage is removed:

• The job finishes in 196.8 seconds, approximately three seconds faster.
• The map output still contains 2,628 partial results.
• The dashboard keeps only one partial result per customer.
• Total displayed revenue falls to 22.7% of the correct value.

Sources. Capture: shuffle simulator 0.1, Python 3.13, 2026-10-14; fixture orders_s0_small.jsonl sha256 5ac4be13… and customers_s0.jsonl sha256 0c688885…; profiles nightly_revenue, lab_cluster (learner pack). The two-stage model is MapReduce’s (J. Dean and S. Ghemawat, “MapReduce: simplified data processing on large clusters”, OSDI 2004), whose combiner and reduce are the page’s map-side state and merge; the numbers are the module’s own.