One Busy Pump Can Keep Every Worker Waiting

Botie can filter each file independently. Now he wants a count for each pump across all files. Most pumps have a few readings, but P7 supplies nearly everything. Several workers finish while the worker responsible for P7 keeps running.

The problem is how the records meet. Adding workers does not divide a group that is still assigned to one worker. We need to see the routing rule before deciding whether more machines, a different join plan or a partial aggregation would help.

The filter worked, but almost all accepted records reached one worker.

A filter can decide whether a reading passes by looking at that reading alone. Counting all readings from P7 is different: P7 may occur in files handled by several workers. Those workers have to combine their contributions before they can give us P7’s total.

The grouping value, P7 here, is called the key. A partition is one portion of the work. A shuffle moves records or partial summaries between partitions so that the contributions for a key can be combined. The movement can take longer than the final addition.

In Spark’s physical plan, an exchange marks a place where that movement is required.

Suppose one worker receives most records. The other workers cannot complete that unsplit group merely by becoming idle. This uneven work distribution is skew. Counts are a first clue; row sizes and the cost of processing each row can also differ.

Look at the slowest tasks and shuffle statistics rather than only the average task time. A long tail can come from a large key, a large record, a spill to disk or another resource problem. Our workshop supplies equal-cost records so routing is the only variable.

Follow one hot pump through the routing rule before deciding whether more partitions would change its destination.

counts = {}
for pump in ["A", "A", "B"]:
    counts[pump] = counts.get(pump, 0) + 1
print(counts)

{'A': 2, 'B': 1}

For a count, split a hot pump into temporary subgroups, count each subgroup, then add counts by the original pump. Adding a small subgroup marker is often called salting. Keep the original key because the marker is an execution detail, not a new pump.

This reasoning relies on a mergeable calculation. Summing counts is valid; averaging subgroup averages without their counts is not. Joins need their own correct treatment of both sides. Arbitrarily salting one join input can lose matches.

A subgroup counts = [2, 2]
A final count = 2 + 2 = 4
B final count = 1

Inspect the existing partial aggregates before adding an extra salting stage.

The reference groups by pump and requests counts. Explain shows the engine’s chosen plan; the bounded display then requests the result. A larger workload can use adaptive execution, but the exact plan depends on Spark and Databricks versions and the available statistics.

Keep a small unpartitioned reference result while changing execution strategy. Performance work has failed if it produces a faster answer for a different grouping rule.

On a real workload, compare the slow-task distribution before and after a justified plan change.

raw = spark.createDataFrame(
    [("A", 0), ("A", 1), ("A", 2), ("A", 3), ("B", 4)],
    "pump string, reading_id long",
)
counts = raw.groupBy("pump").count()
counts.explain("formatted")
counts.orderBy("pump").show()

The workshop adds routingKey to each dictionary. It cycles the hot key through the requested number of salt groups, then routes by a stated character-code hash. Count each resulting worker list, including empty lists.

That hash makes our expected answers inspectable. Spark uses different hashing and can change its physical partitioning. The resulting worker counts are therefore outputs of this small routing model, not predictions of a real Databricks task layout.

The Python challenge returns group counts under the original pump key as well as worker loads. Both matter: better balance and unchanged meaning.

Yesterday P7 produced most of the readings. Today another pump may be busiest. A routing choice should follow the records that actually arrived, so inspect the new key frequencies before reusing yesterday’s hot-key setting. Creating extra groups also adds work that our count-only model leaves out.

Check two results together: how much work reaches the busiest worker, and whether the final counts still describe the original pumps. An evenly divided list is useful only if every observation still contributes to the right total.

Next we will hold the calculation steady and change which version of the readings it receives.