🧹Data wrangling patterns

Most "data engineering" contest problems are one of these patterns in disguise. Learn to recognise them quickly. Each snippet is self-contained and ready to adapt.

P1 · Group & aggregate (several metrics at once)

from collections import defaultdict
stats = defaultdict(lambda: {"n": 0, "sum": 0.0, "min": float("inf"), "max": float("-inf"), "users": set()})
for r in rows:
    s = stats[r["country"]]
    s["n"] += 1; s["sum"] += r["amount"]
    s["min"] = min(s["min"], r["amount"]); s["max"] = max(s["max"], r["amount"])
    s["users"].add(r["user"])                      # COUNT(DISTINCT user)
for k in sorted(stats):
    s = stats[k]
    print(k, s["n"], f'{s["sum"] / s["n"]:.2f}', len(s["users"]))

P2 · Deduplicate / keep latest version (CDC, upserts)

latest = {}
for r in records:                                  # one pass, O(#keys) memory
    k = r["id"]
    if k not in latest or r["updated_at"] >= latest[k]["updated_at"]:   # >= : later line wins ties
        latest[k] = r
# deletes in CDC streams: {"op": "d"} then drop the key
result = [r for r in latest.values() if r.get("op") != "d"]

P3 · Flatten nested JSON (and explode arrays)

def flatten(d, prefix="", sep="."):
    out = {}
    for k, v in d.items():
        key = f"{prefix}{sep}{k}" if prefix else k
        if isinstance(v, dict) and v:
            out.update(flatten(v, key, sep))
        else:
            out[key] = v
    return out

# "explode": one output row per array element (like SQL UNNEST / pandas explode)
order = {"id": 7, "items": [{"sku": "A", "qty": 2}, {"sku": "B", "qty": 1}]}
rows = [{"order_id": order["id"], **item} for item in order.get("items", [])]

P4 · Hash join in Python (inner / left)

users_by_id = {u["id"]: u for u in users}         # build side = smaller table
for o in orders:                                   # probe side = bigger / streamed table
    u = users_by_id.get(o["user_id"])              # left join: u may be None
    print(o["id"], u["name"] if u else "UNKNOWN")

# one-to-many build side: index to a list
from collections import defaultdict
items_by_order = defaultdict(list)
for it in items: items_by_order[it["order_id"]].append(it)

P5 · Sessionization (inactivity gap)

events.sort(key=lambda e: (e["user"], e["ts"]))
sessions, cur = [], None
for e in events:
    if cur and cur["user"] == e["user"] and e["ts"] - cur["end"] <= GAP:
        cur["end"] = e["ts"]; cur["n"] += 1
    else:
        if cur: sessions.append(cur)
        cur = {"user": e["user"], "start": e["ts"], "end": e["ts"], "n": 1}
if cur: sessions.append(cur)                        # don't forget the last one!

P6 · Merge overlapping intervals

Use this for subscriptions, maintenance windows, "total time online" and booking conflicts.

def merge(intervals):                 # [(start, end), ...]
    out = []
    for s, e in sorted(intervals):
        if out and s <= out[-1][1]:     # use < if touching intervals should NOT merge
            out[-1][1] = max(out[-1][1], e)
        else:
            out.append([s, e])
    return out

# max concurrent sessions: sweep line
points = sorted([(s, 1) for s, e in intervals] + [(e, -1) for s, e in intervals])   # (e,-1) sorts before (e,1): end before start
cur = best = 0
for _, d in points:
    cur += d; best = max(best, cur)

P7 · Streaks / gaps & islands

from datetime import date, timedelta
def longest_streak(days):              # days: iterable of date
    days = sorted(set(days))
    best = run = 0; prev = None
    for d in days:
        run = run + 1 if prev and d - prev == timedelta(days=1) else 1
        best = max(best, run); prev = d
    return best

P8 · Cleaning & normalisation

import re, unicodedata
from datetime import datetime

def clean_str(s):
    if s is None: return None
    s = unicodedata.normalize("NFKC", s).strip()
    s = re.sub(r"\s+", " ", s)
    return s or None                       # "" becomes NULL

def to_float(x):
    try: return float(str(x).replace(",", ""))
    except (TypeError, ValueError): return None

FORMATS = ("%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d %H:%M", "%Y-%m-%dT%H:%M:%S")
def parse_dt(s):
    for f in FORMATS:
        try: return datetime.strptime(s.strip(), f)
        except ValueError: pass
    return None                            # quarantine instead of crashing

P9 · Validate & quarantine (data quality)

good, bad = [], []
for i, r in enumerate(rows):
    errs = []
    if not r.get("id"): errs.append("missing id")
    if r.get("amount") is not None and r["amount"] < 0: errs.append("negative amount")
    if r.get("email") and not re.fullmatch(r"[^@\s]+@[^@\s]+\.\w+", r["email"]): errs.append("bad email")
    (bad if errs else good).append((i, r, errs))

The usual data-quality dimensions are completeness (nulls), validity (type, range, regex), uniqueness (PK dupes), consistency (FK exists, totals match), timeliness (freshness) and accuracy.

P10 · Pivot / unpivot in pure Python

pivot = defaultdict(lambda: defaultdict(int))
cols = set()
for r in rows:
    pivot[r["user"]][r["status"]] += 1; cols.add(r["status"])
cols = sorted(cols)
print("user", *cols)
for u in sorted(pivot):
    print(u, *(pivot[u][c] for c in cols))    # missing → 0 (defaultdict)

P11 · Rolling window (moving average over the last N)

from collections import deque
win, s = deque(), 0
for x in values:
    win.append(x); s += x
    if len(win) > N: s -= win.popleft()
    print(f"{s / len(win):.2f}")      # O(1) per step

P12 · Incremental load with a watermark + upsert

def incremental_load(target: dict, source_rows, watermark):
    new_wm = watermark
    for r in source_rows:
        if r["updated_at"] <= watermark:     # already processed
            continue
        target[r["id"]] = r                  # UPSERT (idempotent: rerun → same state)
        new_wm = max(new_wm, r["updated_at"])
    return target, new_wm                    # persist new_wm only AFTER a successful write

P13 · Too big for memory → stream / chunk / partition

# 1) stream line by line (generators)       2) pandas: read_csv(..., chunksize=100_000)
# 3) external sort: sort chunks → write runs → heapq.merge(*runs)
# 4) hash-partition by key into K files → each partition fits in memory → process separately
import heapq
def external_sort(lines, chunk=100_000):
    runs, buf = [], []
    for line in lines:
        buf.append(line)
        if len(buf) == chunk:
            runs.append(sorted(buf)); buf = []    # (in real life: write each run to a temp file)
    runs.append(sorted(buf))
    return heapq.merge(*runs)

P14 · Top-K per group (without pandas)

import heapq
from collections import defaultdict
by_cat = defaultdict(list)
for p in products:
    by_cat[p["category"]].append(p)
for cat in sorted(by_cat):
    top = heapq.nsmallest(3, by_cat[cat], key=lambda p: (-p["sales"], p["name"]))   # sales DESC, name ASC
    print(cat, *(p["name"] for p in top))
# keep ties like RANK(): find the K-th value, then keep everything >= it
#   cutoff = sorted({p["sales"] for p in grp}, reverse=True)[:K][-1]

P15 · Fill the gaps (date spine / zero rows)

from collections import Counter
from datetime import date, timedelta
def daterange(start: date, end: date):          # inclusive
    for i in range((end - start).days + 1):
        yield start + timedelta(days=i)

counts = Counter(r["day"] for r in rows)        # only days that HAVE data
for d in daterange(date(2026, 5, 1), date(2026, 5, 7)):
    print(d.isoformat(), counts.get(d.isoformat(), 0))   # days with no data → 0

"Report every day / hour / month, including empty ones" is a frequent hidden-test trap. A Counter only knows about keys it has seen. Build the full calendar and LEFT-JOIN onto it, in SQL or in Python.

P16 · As-of lookup (value valid at a point in time)

import bisect
rates.sort()                                     # [(ts, rate), ...]
times = [t for t, _ in rates]
def rate_at(ts):
    i = bisect.bisect_right(times, ts) - 1       # last rate with time <= ts
    return rates[i][1] if i >= 0 else None
# SCD2 lookup is the same idea: the row with valid_from <= ts < valid_to

P17 · Parse defensively, quarantine the rest (JSON Lines)

import json, sys
good, bad = [], []
for lineno, line in enumerate(sys.stdin, 1):
    if not line.strip():
        continue
    try:
        rec = json.loads(line)
        if not isinstance(rec, dict):
            raise ValueError("not an object")
        amount = rec["amount"]                    # KeyError if missing
        if isinstance(amount, bool) or not isinstance(amount, (int, float)):
            raise ValueError("amount is not a number")
        good.append(rec)
    except (ValueError, KeyError, TypeError) as e:
        bad.append((lineno, str(e)))             # keep the reason: that's what a DLQ is for