🧠Algorithms that show up in data tasks

Data-engineering contests aren't ICPC, but some problems hide a classic algorithm behind a "pipeline" story. These are the ones that come up.

Complexity budget

Python does roughly 10⁷ simple operations per second. dict/set lookups are O(1) and x in list is O(n).

Topological sort: DAG of pipeline tasks

import heapq
from collections import defaultdict
def topo(deps):                     # deps: task → [prerequisites]
    nodes = set(deps) | {d for ds in deps.values() for d in ds}
    indeg = {n: 0 for n in nodes}; out = defaultdict(list)
    for t, ds in deps.items():
        for d in ds: out[d].append(t); indeg[t] += 1
    ready = sorted(n for n in nodes if indeg[n] == 0); heapq.heapify(ready)
    order = []
    while ready:
        n = heapq.heappop(ready); order.append(n)
        for c in out[n]:
            indeg[c] -= 1
            if indeg[c] == 0: heapq.heappush(ready, c)
    return order if len(order) == len(nodes) else None   # None means there's a cycle
# Variants: "levels" (tasks that can run in parallel) = BFS layers;
# "minimum total time with unlimited workers" = longest path with durations (critical path).

Union-Find: entity resolution / identity stitching

Use this for "users are the same person if they share an email OR phone OR device; count the unique people".

parent = {}
def find(x):
    parent.setdefault(x, x)
    while parent[x] != x:
        parent[x] = parent[parent[x]]    # path halving
        x = parent[x]
    return x
def union(a, b):
    ra, rb = find(a), find(b)
    if ra != rb: parent[max(ra, rb)] = min(ra, rb)   # deterministic root = smallest

for acc in accounts:                      # link the account to each of its identifiers
    for ident in (acc["email"], acc["phone"]):
        if ident: union(("acc", acc["id"]), ("id", ident))
people = {find(("acc", a["id"])) for a in accounts}

Hashing, two pointers, binary search

# merge-join two SORTED streams (like a sort-merge join) in O(n + m)
def merge_join(a, b):                    # lists of (key, val) sorted by key, unique keys
    i = j = 0; out = []
    while i < len(a) and j < len(b):
        if a[i][0] == b[j][0]: out.append((a[i][0], a[i][1], b[j][1])); i += 1; j += 1
        elif a[i][0] < b[j][0]: i += 1
        else: j += 1
    return out

# "as-of" lookup: the exchange rate valid at time t (like pandas.merge_asof)
import bisect
times = [r[0] for r in rates]            # sorted
idx = bisect.bisect_right(times, t) - 1  # the last rate with time <= t
rate = rates[idx][1] if idx >= 0 else None

Prefix sums, sliding window, sweep line

from itertools import accumulate
pre = [0, *accumulate(values)]           # sum(values[l:r]) = pre[r] - pre[l]  in O(1)

# longest window where the distinct-key count ≤ K (two pointers)
from collections import Counter
cnt, l, best = Counter(), 0, 0
for r, x in enumerate(arr):
    cnt[x] += 1
    while len(cnt) > K:
        cnt[arr[l]] -= 1
        if cnt[arr[l]] == 0: del cnt[arr[l]]
        l += 1
    best = max(best, r - l + 1)

Streaming / approximate structures (interview talk)

• Reservoir sampling: a uniform sample of k items from a stream of unknown length.
• HyperLogLog: approximate COUNT DISTINCT in KBs of memory (approx_count_distinct in Spark).
• Count-Min Sketch: approximate frequencies / heavy hitters. Bloom filter: "definitely not in set" membership tests.
• Consistent hashing: assigns keys to shards with minimal movement when nodes change.

import random
def reservoir(stream, k):
    sample = []
    for i, x in enumerate(stream):
        if i < k: sample.append(x)
        else:
            j = random.randint(0, i)
            if j < k: sample[j] = x
    return sample

Caches and rate limiters

from collections import OrderedDict, deque

class LRU:                                   # O(1) get/put
    def __init__(self, cap): self.cap, self.d = cap, OrderedDict()
    def get(self, k):
        if k not in self.d: return None
        self.d.move_to_end(k); return self.d[k]
    def put(self, k, v):
        self.d[k] = v; self.d.move_to_end(k)
        if len(self.d) > self.cap: self.d.popitem(last=False)   # evict least recently used

def allowed(log: deque, now, limit=100, window=60):   # sliding-window rate limiter per client
    while log and log[0] <= now - window:   # drop timestamps that left the window
        log.popleft()
    if len(log) < limit:
        log.append(now); return True
    return False

Graphs: BFS levels and shortest paths

from collections import deque
import heapq

def bfs_levels(adj, start):                  # unweighted: number of hops (lineage depth, "degrees of separation")
    dist = {start: 0}; q = deque([start])
    while q:
        u = q.popleft()
        for v in adj.get(u, ()):
            if v not in dist:
                dist[v] = dist[u] + 1; q.append(v)
    return dist

def dijkstra(adj, start):                    # weighted, non-negative: adj[u] = [(v, w), ...]
    dist = {start: 0}; pq = [(0, start)]
    while pq:
        d, u = heapq.heappop(pq)
        if d > dist.get(u, float("inf")): continue     # stale entry
        for v, w in adj.get(u, ()):
            nd = d + w
            if nd < dist.get(v, float("inf")):
                dist[v] = nd; heapq.heappush(pq, (nd, v))
    return dist

A pipeline's critical path (minimum total runtime with unlimited workers) is the longest path in the DAG: process tasks in topological order and set finish[t] = duration[t] + max(finish[d] for d in deps[t]).

Binary search on the answer

When a question asks for "the minimum X such that something is feasible" and feasibility is monotonic, binary-search X. Example: the smallest batch size so that all files are processed in at most H hours.

def min_feasible(lo, hi, ok):                # smallest x in [lo, hi] with ok(x) True; ok must be monotonic
    while lo < hi:
        mid = (lo + hi) // 2
        if ok(mid): hi = mid
        else: lo = mid + 1
    return lo

# e.g. minimum workers so that jobs (in order) finish within T, each worker taking consecutive jobs
def fits(workers, jobs, T):
    used, cur = 1, 0
    for j in jobs:
        if j > T: return False
        if cur + j > T: used, cur = used + 1, 0
        cur += j
    return used <= workers

Complexity cheat sheet for Python built-ins