Data that does not fit: external sort, streams

Some data never fits in memory: a day of logs larger than the machine’s RAM, or a stream of events that never ends. Algorithms that load everything first fail on both, and two families of techniques take over. External algorithms work through files in a few sequential passes; the most important is external merge sort, which sorts memory-sized chunks into sorted files called runs and then merges all the runs at once with a heap. Streaming algorithms read each item once and keep a summary whose size does not grow with the stream: exact counters while the keys fit in memory, and approximate summaries, random samples and sketches, when they do not. This lesson runs both on real files and generated streams.

External merge sort in two phases

Suppose a 200 MB log must be sorted by time on a machine that can spare 20 MB. External merge sort does it in two phases:

1. Runs. Read 20 MB of lines, sort them in memory and write them to a file, a sorted run. Repeat until the input ends: 10 runs.
2. Merge. Open all 10 runs and keep the next unwritten line of each in a min-heap of 10 entries. Write the smallest line, then refill the heap from the run it came from. When every run is used up, the output is sorted.

Each phase reads all the data once and writes it once, so the sort moves 4 × 200 MB = 800 MB through the disk, whatever the lines say. Disks and SSDs read long sequential stretches far faster than scattered blocks, and sorting in memory is quick next to that, so the number of passes over the data decides the time. The merge holds only one line of each run, plus a read buffer per file, so its memory grows with the number of runs, not with the size of the data. Python’s heapq.merge performs phase 2: its documentation says it does not pull the data into memory all at once.

# External merge sort on a small scale: sort a log file of 50,000 lines while holding at most 5,000 lines in memory.
# Phase 1 cuts the file into sorted runs; phase 2 merges the runs with a heap (heapq.merge reads them lazily).
import heapq
import itertools
import os
import random

LINES = 50_000
BUDGET = 5_000  # the most lines phase 1 holds in memory at once


def make_log(path, n, seed):
    """A synthetic access log: the millisecond of the day, the service and a request id."""
    rng = random.Random(seed)
    services = ["auth", "cart", "pay", "search", "ship"]
    with open(path, "w") as f:
        for _ in range(n):
            f.write(f"{rng.randrange(86_400_000):08d} {rng.choice(services):6} req{rng.randrange(10**6):06d}\n")


def make_runs(path, budget):
    """Phase 1: read `budget` lines at a time, sort them in memory, write each batch as a run file."""
    runs = []
    with open(path) as f:
        while chunk := list(itertools.islice(f, budget)):
            chunk.sort()
            name = f"run{len(runs):03d}.txt"
            with open(name, "w") as out:
                out.writelines(chunk)
            runs.append(name)
    return runs


def merge_runs(runs, out_path):
    """Phase 2: one k-way merge; heapq.merge keeps one line of each run in a heap."""
    files = [open(name) for name in runs]
    try:
        with open(out_path, "w") as out:
            out.writelines(heapq.merge(*files))
    finally:
        for f in files:
            f.close()


make_log("access.log", LINES, seed=3)
size = os.path.getsize("access.log")
runs = make_runs("access.log", BUDGET)
run_bytes = sum(os.path.getsize(r) for r in runs)
merge_runs(runs, "sorted.log")

with open("access.log") as f:
    expected = sorted(f)  # only to check the result: this small file fits in memory
with open("sorted.log") as f:
    result = f.readlines()
print(f"input: {LINES:,} lines, {size:,} bytes")
print(f"phase 1: {len(runs)} runs of at most {BUDGET:,} lines, each sorted in memory")
print(f"phase 2: one {len(runs)}-way merge, holding one line of each run in a heap")
print(f"bytes read: {size + run_bytes:,} (the data twice); bytes written: {run_bytes + os.path.getsize('sorted.log'):,}")
print(f"first lines: {result[0].strip()} | {result[1].strip()}")
print(f"output sorted and complete: {result == expected}")
for name in runs + ["access.log", "sorted.log"]:
    os.remove(name)

input: 50,000 lines, 1,300,000 bytes
phase 1: 10 runs of at most 5,000 lines, each sorted in memory
phase 2: one 10-way merge, holding one line of each run in a heap
bytes read: 2,600,000 (the data twice); bytes written: 2,600,000
first lines: 00002015 cart   req217182 | 00002550 pay    req798609
output sorted and complete: True

The run sorted 50,000 lines while never holding more than 5,000 of them, read the data twice and wrote it twice, and the result matches Python’s sorted() line for line. With 10 runs the heap had 10 entries; a file of any size would give the same shape, with more or longer runs.

Practise: merge k sorted streams

import heapq


def merge_sorted(inputs):
    """Yield the items of the sorted iterators in `inputs` in sorted order, reading each one lazily."""
    return iter([])

from itertools import count, islice

from merge import merge_sorted


def test_three_inputs():
    """merges three sorted lists, one of them empty"""
    assert list(merge_sorted([iter([1, 4, 9]), iter([2, 3, 10]), iter([])])) == [1, 2, 3, 4, 9, 10]


def test_no_inputs():
    """no inputs give no output"""
    assert list(merge_sorted([])) == []


def test_duplicates_and_negatives():
    """keeps every copy of a repeated item"""
    assert list(merge_sorted([iter([-5, 0, 0, 7]), iter([0, 7, 7])])) == [-5, 0, 0, 0, 7, 7, 7]


def test_strings():
    """works for anything that can be compared, such as log lines"""
    out = merge_sorted([iter(["0815 pay", "0930 cart"]), iter(["0816 auth"]), iter(["0001 ship", "0999 pay"])])
    assert list(out) == ["0001 ship", "0815 pay", "0816 auth", "0930 cart", "0999 pay"]


def test_equal_items_in_input_order():
    """an item of an earlier input comes before an equal item of a later one"""
    a, b = [1.0, 2.0], [1, 2]  # 1 == 1.0, but the two can be told apart by type
    out = list(merge_sorted([iter(a), iter(b)]))
    assert out == [1.0, 1, 2.0, 2] and [type(x) for x in out] == [float, int, float, int]


def test_lazy_with_endless_inputs():
    """reads only what it needs: these inputs never end"""
    merged = merge_sorted([count(0, 3), count(1, 3), count(2, 3)])
    assert list(islice(merged, 12)) == list(range(12))


def test_holds_little_of_each_input():
    """has read at most two items of an input when it yields the first item"""
    pulled = [0, 0]

    def tracked(items, i):
        for x in items:
            pulled[i] += 1
            yield x

    merged = merge_sorted([tracked([1, 2, 3, 4], 0), tracked([10, 20], 1)])
    assert next(merged) == 1
    assert pulled[0] <= 2 and pulled[1] == 1
    assert list(merged) == [2, 3, 4, 10, 20]

/** Yield the items of the sorted iterables in `inputs` in sorted order, reading each one lazily. */
export function* mergeSorted(inputs) {
  yield* [];
}

import { test, assert } from 'mysmartcopilot:test';
import { mergeSorted } from './merge.mjs';

function* countFrom(start, step) {
  for (let x = start; ; x += step) yield x;
}
function take(iterator, n) {
  const out = [];
  for (const x of iterator) {
    if (out.length === n) break;
    out.push(x);
  }
  return out;
}

test('merges three sorted lists, one of them empty', () => {
  assert.deepEqual([...mergeSorted([[1, 4, 9], [2, 3, 10], []])], [1, 2, 3, 4, 9, 10]);
});
test('no inputs give no output', () => {
  assert.deepEqual([...mergeSorted([])], []);
});
test('keeps every copy of a repeated item', () => {
  assert.deepEqual([...mergeSorted([[-5, 0, 0, 7], [0, 7, 7]])], [-5, 0, 0, 0, 7, 7, 7]);
});
test('works for anything that can be compared, such as log lines', () => {
  const out = [...mergeSorted([['0815 pay', '0930 cart'], ['0816 auth'], ['0001 ship', '0999 pay']])];
  assert.deepEqual(out, ['0001 ship', '0815 pay', '0816 auth', '0930 cart', '0999 pay']);
});
test('reads only what it needs: these inputs never end', () => {
  const merged = mergeSorted([countFrom(0, 3), countFrom(1, 3), countFrom(2, 3)]);
  assert.deepEqual(take(merged, 12), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
});
test('has read at most two items of an input when it yields the first item', () => {
  const pulled = [0, 0];
  function* tracked(items, i) {
    for (const x of items) {
      pulled[i]++;
      yield x;
    }
  }
  const merged = mergeSorted([tracked([1, 2, 3, 4], 0), tracked([10, 20], 1)]);
  assert.equal(merged.next().value, 1);
  assert.ok(pulled[0] <= 2 && pulled[1] === 1);
  assert.deepEqual([...merged], [2, 3, 4, 10, 20]);
});

One pass, bounded memory

A stream is data you see once, in order, as it arrives: requests to a server, sensor readings, clicks. A streaming algorithm updates a small summary for each item and can answer at any moment. Some questions have exact one-pass answers in constant memory: the count, the sum and the mean, the minimum and the maximum, and the variance with Welford’s method, which updates a running mean and a running sum of squared differences.

# One pass over a stream of a million response times, keeping five numbers: count, mean, the sum of squared
# differences from the mean (Welford's method, for the standard deviation), minimum and maximum. The stream is a
# generator, so no time is ever stored, and the memory used does not grow with the stream.
import math
import random


def response_times(n, seed):
    """A stream of n response times in milliseconds: at least 5 ms, mostly fast, with a slow tail."""
    rng = random.Random(seed)
    for _ in range(n):
        yield 5 + rng.expovariate(1 / 40) + (rng.random() < 0.02) * rng.uniform(500, 2000)


count = 0
mean = 0.0
m2 = 0.0  # the sum of squared differences from the current mean
low = math.inf
high = -math.inf
for t in response_times(1_000_000, seed=5):
    count += 1
    delta = t - mean
    mean += delta / count
    m2 += delta * (t - mean)
    low = min(low, t)
    high = max(high, t)

print(f"times seen: {count:,}, numbers kept: 5")
print(f"mean {mean:.2f} ms, standard deviation {math.sqrt(m2 / (count - 1)):.2f} ms")
print(f"fastest {low:.2f} ms, slowest {high:,.1f} ms")

times seen: 1,000,000, numbers kept: 5
mean 70.24 ms, standard deviation 190.24 ms
fastest 5.00 ms, slowest 2,287.6 ms

Other questions need memory that grows with the data if the answer must be exact. The number of distinct users needs every user seen so far, the most frequent terms need a counter for every term, and the median needs every value. When that memory is not available, a summary of fixed size answers approximately:

• Exact counters and hash maps give exact answers, with memory that grows with the number of distinct keys.
• Reservoir sampling keeps a uniform random sample of k items from a stream of unknown length.
• The Misra–Gries algorithm keeps k − 1 counters and never loses an item that makes up more than 1/k of the stream.
• A count-min sketch estimates how often any item appeared from a fixed table of counters, and is never too low.
• A Bloom filter answers “seen before?” in a few bits per item, with false positives but no false negatives.
• HyperLogLog estimates the number of distinct items from a few kilobytes, typically within 1.04/√m for m registers.

Check yourself

External merge sort at full size, with numbers Pro: opens with a Pro, Premium or Ultimate pass

Streaming summaries measured: samples, heavy hitters, distinct counts Pro: opens with a Pro, Premium or Ultimate pass

Interview questions

Key takeaways

• External merge sort writes memory-sized sorted runs, then merges them with a heap holding one item per run; each pass reads and writes all the data, so passes decide the cost.
• A merge needs memory for one buffer per run, not for the data, and heapq.merge reads its inputs lazily.
• Streaming algorithms see each item once and keep a summary that does not grow with the stream: exact when the answer allows it, such as counts, means and extremes.
• When exact state does not fit, samples and sketches answer approximately from a fixed size, each with a stated error.