Aggregation pipeline
Before you model data you have to understand it — and that means grouping, counting, summing, averaging. A pure-Python aggregation pipeline is three small stages chained together:
- Group-by — partition a list of records into a
dictkeyed by some field's value, each key mapping to the list of records that share it. Acollections.defaultdict(list)makes this clean:groups[key].append(record)with no "is the key there yet?" check. - Aggregate — collapse each group's list down to one value with a function (
sum,len, a mean…). A dict comprehension overgroups.items()does it in a line. - Sort / top-N — order the aggregated results, usually by value descending, and optionally keep the top few.
sorted(d.items(), key=lambda kv: kv[1], reverse=True).
Keeping the aggregation function as a parameter (agg_fn) makes one pipeline do sum, count, mean, max — whatever you pass in. That's the same flexibility groupby().agg() gives you.
*In production this is pandas.DataFrame.groupby(). What's below is, near enough, what pandas runs under the hood.*
Syntax
from collections import defaultdict
def group_by(records, key):
groups = defaultdict(list)
for r in records:
groups[r[key]].append(r)
return dict(groups)
def aggregate(groups, agg_fn): # agg_fn: list[record] -> value
return {k: agg_fn(recs) for k, recs in groups.items()}
def top_n(agg_result, n):
return sorted(agg_result.items(), key=lambda kv: kv[1], reverse=True)[:n]Worked examples
Group records by a field
from collections import defaultdict
def group_by(records, key):
groups = defaultdict(list)
for r in records:
groups[r[key]].append(r)
return dict(groups)
txns = [
{"region": "North", "revenue": 4200},
{"region": "South", "revenue": 1800},
{"region": "North", "revenue": 3100},
{"region": "East", "revenue": 950},
{"region": "South", "revenue": 2200},
]
groups = group_by(txns, "region")
print({k: len(v) for k, v in groups.items()})
# Output:
# {'North': 2, 'South': 2, 'East': 1}Aggregate each group — sum, count, mean (agg_fn is a parameter)
import statistics
from collections import defaultdict
def group_by(records, key):
groups = defaultdict(list)
for r in records:
groups[r[key]].append(r)
return dict(groups)
def aggregate(groups, agg_fn):
return {k: agg_fn(recs) for k, recs in groups.items()}
txns = [
{"region": "North", "revenue": 4200},
{"region": "South", "revenue": 1800},
{"region": "North", "revenue": 3100},
{"region": "East", "revenue": 950},
{"region": "South", "revenue": 2200},
]
g = group_by(txns, "region")
print(aggregate(g, lambda recs: sum(r["revenue"] for r in recs)))
print(aggregate(g, len))
print(aggregate(g, lambda recs: round(statistics.mean(r["revenue"] for r in recs), 1)))
# Output:
# {'North': 7300, 'South': 4000, 'East': 950}
# {'North': 2, 'South': 2, 'East': 1}
# {'North': 3650, 'South': 2000, 'East': 950}Sort + take the top N
def top_n(agg_result, n):
return sorted(agg_result.items(), key=lambda kv: kv[1], reverse=True)[:n]
totals = {"North": 7300, "South": 4000, "East": 950}
print(top_n(totals, 2))
# Output:
# [('North', 7300), ('South', 4000)]Common mistakes & gotchas
Plain dict needs an init; defaultdict doesn't
With a normal dict, groups[key].append(r) raises KeyError the first time a key is seen — you must do if key not in groups: groups[key] = [] first. defaultdict(list) creates the empty list automatically. Either works; mixing them up is the bug.
sorted() needs a key for tuples
sorted(d.items()) sorts by the KEY (the first tuple element), not the value. To rank by aggregated value you must pass key=lambda kv: kv[1]. Forgetting it sorts alphabetically by group name — a silent wrong answer, not a crash.
Iterators are single-use
If agg_fn is sum(r['revenue'] for r in recs), that generator is consumed once — fine here. But if you tried to iterate the same generator twice inside one agg_fn (e.g. mean as sum/len from one generator), the second pass sees nothing. Materialise to a list first if you need it twice.
Why it matters
Group-by-aggregate-sort is the single most common shape in all of data work — every dashboard, report, and exploratory analysis is built from it. Owning the hand-rolled version means you understand exactly what pandas is doing, and you can do it even in an environment where pandas isn't available.