Data Pipeline Patterns

Group Counts

Grouping turns individual records into counts keyed by a shared field.

grouping records A grouping stage updates one dictionary entry for each record key.

Group Counts

status_filter
group_counts.py
Replay: real traced execution (multi-file project)
orders = [
    {"region": "east", "status": "open"},
    {"region": "west", "status": "open"},
    {"region": "west", "status": "closed"},
]

status_filter = "all"

counts = {}
for order in orders:
    if status_filter != "all" and order["status"] != status_filter:
        continue
    region = order["region"]
    counts[region] = counts.get(region, 0) + 1

parts = []
for region in sorted(counts):
    parts.append(region + "=" + str(counts[region]))

print("counts=" + ";".join(parts))

orders = [
    {"region": "east", "status": "open"},
    {"region": "west", "status": "open"},
    {"region": "west", "status": "closed"},
]

status_filter = "open"

counts = {}
for order in orders:
    if status_filter != "all" and order["status"] != status_filter:
        continue
    region = order["region"]
    counts[region] = counts.get(region, 0) + 1

parts = []
for region in sorted(counts):
    parts.append(region + "=" + str(counts[region]))

print("counts=" + ";".join(parts))

  1. orders ← [{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}]

    1orders→ [{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}] = [2    {"region": "east", "status": "open"},3    {"region": "west", "status": "open"},4    {"region": "west", "status": "closed"},5]67status_filter→ all = "all"  #@status_filter="open"89counts→ {} = {}10for order in orders:

  2. region ← east, counts ← {'east': 1}, counts[region] ← 1

    pass 1 of 3
    9counts = {}10for order{'region': 'east', 'status': 'open'} in orders[{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}]:11    if status_filter != "all" and order["status"] != status_filter:12        continue13    region→ east = order["region"]east14    counts[region]→ 1 = counts→ {'east': 1}.get(regioneast, 0) + 1
    All 3 passes — pass 1 is the card above
    passorderorder[”region”]regioncountscounts[region]
    1{'region': 'east', 'status': 'open'}easteast{} → {'east': 1}1
    2{'region': 'west', 'status': 'open'}westwest{'east': 1} → {'east': 1, 'west': 1}1
    3{'region': 'west', 'status': 'closed'}westwest{'east': 1, 'west': 1} → {'east': 1, 'west': 2}2

  3. parts ← []

    16parts→ [] = []17for region in sorted(counts):

  4. parts ← ['east=1']

    pass 1 of 2
    16parts = []17for regioneast in sorted(counts{'east': 1, 'west': 2}):18    parts→ ['east=1'].append(regioneast + "=" + str(counts[region]1))

  5. parts ← ['east=1', 'west=2']

    pass 2 of 2
    16parts = []17for regionwest in sorted(counts{'east': 1, 'west': 2}):18    parts→ ['east=1', 'west=2'].append(regionwest + "=" + str(counts[region]2))

  6. print("counts=" + ";".join(parts))

    20print("counts=" + ";".join(parts['east=1', 'west=2']))
    outputcounts=east=1;west=2

  1. orders ← [{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}]

    1orders→ [{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}] = [2    {"region": "east", "status": "open"},3    {"region": "west", "status": "open"},4    {"region": "west", "status": "closed"},5]67status_filter→ open = "open"89counts→ {} = {}10for order in orders:

  2. region ← east, counts ← {'east': 1}, counts[region] ← 1

    pass 1 of 3
    9counts = {}10for order{'region': 'east', 'status': 'open'} in orders[{'region': 'east', 'status': 'open'}, {'region': 'west', 'status': 'open'}, {'region': 'west', 'status': 'closed'}]:11    if status_filter != "all" and order["status"] != status_filter:12        continue13    region→ east = order["region"]east14    counts[region]→ 1 = counts→ {'east': 1}.get(regioneast, 0) + 1
    All 3 passes — pass 1 is the card above
    passorderorder[”region”]status_filterorder[”status”]regioncountscounts[region]
    1{'region': 'east', 'status': 'open'}east——east{} → {'east': 1}1
    2{'region': 'west', 'status': 'open'}west——west{'east': 1} → {'east': 1, 'west': 1}1
    3{'region': 'west', 'status': 'closed'}—openclosed———

  3. if status_filter != "all" and order["status"] != status_filter:

    10for order in orders:11    if status_filteropen != "all" and order["status"]closed != status_filter:12        continue13    region = order["region"]

  4. parts ← []

    16parts→ [] = []17for region in sorted(counts):

  5. parts ← ['east=1']

    pass 1 of 2
    16parts = []17for regioneast in sorted(counts{'east': 1, 'west': 1}):18    parts→ ['east=1'].append(regioneast + "=" + str(counts[region]1))

  6. parts ← ['east=1', 'west=1']

    pass 2 of 2
    16parts = []17for regionwest in sorted(counts{'east': 1, 'west': 1}):18    parts→ ['east=1', 'west=1'].append(regionwest + "=" + str(counts[region]1))

  7. print("counts=" + ";".join(parts))

    20print("counts=" + ";".join(parts['east=1', 'west=1']))
    outputcounts=east=1;west=1