Basic DE Applications

In a pipeline, top-K rarely arrives as a tidy list of numbers. It arrives as events that must be summed by key or ranked within groups, and often the ranking runs on many partitions whose results have to be merged. Each shape reuses the capped heap and adds a correctness rule of its own.

Aggregate first, then rank

Top 2 customers by revenue, computed from order events, takes 2 steps. A dict sums each customer's orders in a single O(n) pass, then a heap ranks the m distinct customers in O(m log k). Ranking the raw events instead would return c3 and c4 because they placed the 2 largest single orders, which answers a different question.

import heapq
from collections import defaultdict

orders = [
    ("c1", 120.0), ("c2", 45.5), ("c3", 300.0),
    ("c1", 250.0), ("c4", 300.0), ("c2", 210.0),
]
revenue = defaultdict(float)
for customer, amount in orders:
    revenue[customer] += amount

top2 = heapq.nsmallest(2, revenue.items(), key=lambda kv: (-kv[1], kv[0]))
print(top2)

Customers c3 and c4 tie at 300.0 for second place. The key, negated total then customer ID, hands the place to c3, so a rerun that reads the orders in a different sequence prints the same answer. With heapq.nlargest(2, revenue.items(), key=lambda kv: kv[1]), the tie between c3 and c4 falls back to the dict's insertion order. That order follows the sequence the orders were read in, so 2 runs over the same orders can publish different rankings.

When totals keep changing

The capped heap assumes each key arrives once with its final value. In a running aggregation, c2 would enter at 45.5 and climb to 255.5 when its second order lands, while the heap keeps the stale 45.5 entry with no way to find and update it. For a batch, rank once after the sums finish. For an hourly refresh, keep running totals in the dict and rebuild the top-K from it with nsmallest each hour, at O(m log k) per refresh.

Top-K per group: a dict of heaps

Finding the 2 slowest queries in each region takes 1 heap per group. A dict maps each region to its own list, setdefault creates that list the first time a region appears, and every event runs the push-or-replace step against its region's heap. Memory is k entries per group, so 50 regions at k = 2 hold 100 entries however long the log runs.

import heapq

events = [
    ("us", "q1", 340), ("eu", "q2", 120), ("us", "q3", 90),
    ("us", "q4", 510), ("eu", "q5", 800), ("eu", "q6", 60),
    ("us", "q7", 420),
]
k = 2
slowest = {}
for region, query_id, ms in events:
    heap = slowest.setdefault(region, [])
    if len(heap) < k:
        heapq.heappush(heap, (ms, query_id))
    elif ms > heap[0][0]:
        heapq.heapreplace(heap, (ms, query_id))

for region in sorted(slowest):
    print(region, sorted(slowest[region], reverse=True))

eu [(800, 'q5'), (120, 'q2')]
us [(510, 'q4'), (420, 'q7')]

The eu heap rejects q6 at 60 ms because its root is already 120. The us heap fills at q3. After that, q4 evicts the root at the time, which is 90, and q7 evicts the next root, 340. Iterating over sorted(slowest) keeps the region order fixed from run to run.

Merging top-K across partitions

A distributed job computes a local top-K on each partition and merges the results. The global top-K is guaranteed to lie inside the union of the local results only when each key's full total lives on a single partition, which is what partitioning by that key provides. Partition by time instead and each total is split, so the merge can miss the true leader.

Merging the local results sees alice at 100 and bob at 90 ahead of carol at 85 and dave at 70, and reports alice and bob. The true totals put carol first at 165, ahead of alice at 100 and bob at 95, with dave last at 70, so the actual leader is missing: carol's 80 in hour 1 was cut by that partition's top 2. The fix is to shuffle rows so each customer lands on a single partition and gets summed there. Only after that does each partition take its local top-K.

Where else the pattern runs

• Slowest queries per day for an on-call report, ranked by duration
• Largest partitions of a table by row count, to spot skew before a join
• Most frequent error messages in a batch of log lines
• Next task to run in a scheduler, from a heap of (run_at, seq, task)
• Top sellers per category for a daily product feed, using a dict of heaps

About this lesson

This section is part of the Heap & Top-K: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.