Chunked Processing

The previous lesson narrowed the memory cost of a column by ratios of up to 31 times, but it also said that narrowing has a floor: 3,199 rows fit with every type, and as the dataset itself grows no type is enough. This lesson measures what is done when that floor is exceeded — the dataset is not taken into memory at once; it is read chunk by chunk.

The promise of chunking is clear: only one chunk sits in memory at a time, and the peak usage is set by the chunk size. The measure of the lesson reads the price of this promise with two numbers. The first is the gain: by how many times the memory peak shrinks when the chunk size drops. The second is the carried state: the item the operation keeps outside the chunk and that does not shrink when the chunk size changes. This is what determines whether an operation can be computed chunk by chunk.

• SP6. The dataset is the measurement network dataset of the previous lesson: 3,199 rows, 1,329 subscribers, three periods, a total of 66,667.9 m³. It is produced with the same seed and the output is identical from run to run. It is a model.
• SP7. The file carries the dataset’s five fields; the tariff tier is not included in the export because it is derived from the reading.
• SP8. The file is written in period order: first the 1,329 rows of 2024-01, then the 1,035 rows of 2024-02, then the 835 rows of 2024-03. The rows of one subscriber are hundreds of rows apart.
• SP9. The memory peak is the bytes of the rows held in memory at the same time plus the operation state; it is measured with sys.getsizeof and belongs to a runtime with eight-byte pointers. What does not change is the ratios.
• SP10. For two means to count as equal the threshold is 0.05 m³, because readings are reported with one decimal and a difference smaller than that can come only from rounding.

Chunk Size and Memory Peak

# dataset.py — the same measurement network dataset is written to a file. MODEL:
# a generator that produces the same structure was written in place of a real
# export; the seed is visible and the output is identical from run to run.
# 3199 rows, 1329 subscribers, three periods.
import csv, itertools, os, sys, tempfile
SEED = 1246
REGION = ("north", "east", "center", "south", "west")
PERIODS = (("2024-01", 1329), ("2024-02", 1035), ("2024-03", 835))
DENSITY = (0, 1, 2, 2, 2, 3, 4, 2, 2, 1)   # center-weighted measurement network
HEADER = ("subscriber_id", "meter_id", "period", "region", "reading_m3")


def rng(seed: int):
    s = seed & 0xFFFFFFFF

    def next_value() -> float:
        nonlocal s
        s = (1103515245 * s + 12345) & 0xFFFFFFFF
        return s / 4294967296
    return next_value


def records() -> list[tuple]:
    r, out = rng(SEED), []
    for period, size in PERIODS:
        for i in range(size):
            out.append((10000 + i, f"S-{4000 + i}", period,
                        REGION[DENSITY[i % 10]], round(5 + r() * 32.0, 1)))
    return out


r0 = records()
DIR = tempfile.mkdtemp(prefix="measurement-")
PATH = os.path.join(DIR, "readings.csv")
with open(PATH, "w", newline="") as f:
    w = csv.writer(f, lineterminator="\n")
    w.writerow(HEADER)
    w.writerows(r0)

nbytes = lambda p: sys.getsizeof(p) + sum(
    sys.getsizeof(s) + sum(sys.getsizeof(a) for a in s) for s in p)
FULL = nbytes([list(map(str, s)) for s in r0])
print(f"seed {SEED}; file {os.path.getsize(PATH)} bytes, {len(r0)} data rows")
print(f"if the whole dataset is taken into memory {FULL} bytes, "
      f"{round(FULL / len(r0))} bytes per row")
print(f"total {sum(x[4] for x in r0):.1f} m3, distinct subscribers {len({x[0] for x in r0})}")

seed 1246; file 102557 bytes, 3199 data rows
if the whole dataset is taken into memory 1152229 bytes, 360 bytes per row
total 66667.9 m3, distinct subscribers 1329

The file takes 102,548 bytes on disk and 1,152,229 bytes in memory — eleven times. The difference is the subject of the previous lesson: once a text line is parsed, every field becomes a separate object. This lesson takes that multiplier as data and asks a single question: is it possible to produce the same numbers without holding the whole dataset in memory at the same time.

# chunk.py — the file is read chunk by chunk. The total, the row count, the maximum and the
# per-region total are accumulated in the same pass; the chunk size changes only the
# number of rows held in memory at the same time.
def chunks(path: str, n: int):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)                                    # header row
        while (p := list(itertools.islice(reader, n))):
            yield p


def single_pass(path: str, n: int) -> dict:
    peak = num_chunks = count = 0
    total, largest, by_region = 0.0, 0.0, {}
    for p in chunks(path, n):
        peak, num_chunks = max(peak, nbytes(p)), num_chunks + 1
        for s in p:
            o = float(s[4])
            total, count, largest = total + o, count + 1, max(largest, o)
            by_region[s[3]] = by_region.get(s[3], 0.0) + o
    return {"chunks": num_chunks, "peak": peak, "count": count, "total": round(total, 1),
            "largest": largest, "region": round(max(by_region.values()), 1)}


G = (14, 10, 12, 11, 8, 13, 16)
fmt = lambda h: "".join(str(v).rjust(G[j]) if j else str(v).ljust(G[0])
                        for j, v in enumerate(h))
print(fmt(["chunk size", "chunks", "peak bytes", "peak/full", "passes", "total m3",
           "largest region"]))
for n in (100, 400, 1600, 3199):
    r = single_pass(PATH, n)
    print(fmt([n, r["chunks"], r["peak"], f"{r['peak'] / FULL:.3f}", 1, r["total"],
               r["region"]]))
print(f"at chunk size 100 the peak is {round(FULL / single_pass(PATH, 100)['peak'])} times "
      f"smaller than holding the whole dataset; in all four runs the total is "
      f"{single_pass(PATH, 100)['total']} m3 and the rows {single_pass(PATH, 100)['count']}")

chunk size        chunks  peak bytes  peak/full  passes     total m3  largest region
100                   32       36132      0.031       1      66667.9         33462.3
400                    8      144085      0.125       1      66667.9         33462.3
1600                   2      577624      0.501       1      66667.9         33462.3
3199                   1     1152229      1.000       1      66667.9         33462.3
at chunk size 100 the peak is 32 times smaller than holding the whole dataset; in all four runs the total is 66667.9 m3 and the rows 3199

The gain is linear and can be read from the table: when the chunk size drops from 3,199 to 100, the peak falls from 1,152,229 bytes to 36,132 bytes, to one thirty-second of the dataset. The number of passes is one in all four runs; the file is read once. The total also comes out the same in all four runs — 66,667.9 m³ — and the total of the largest region stays fixed at 33,462.3 m³. These four operations are indifferent to chunk size: each keeps a constant number of accumulators, and only those accumulators pass from chunk to chunk.

The cost of chunk size is not visible in the table, because this lesson does not measure time. A smaller chunk means more calls, a larger chunk means more memory; the choice lies between these two numbers, and the peak limit comes from the hardware. The side that can be measured is the peak bytes, and it comes out proportional to the chunk size.

Which Operation Finishes Within the Chunk

# operation.py — which operation finishes within the chunk and which carries state outside it.
# Measure: the state items and bytes carried, independent of chunk size.
THRESHOLD = 0.05   # readings are reported with one decimal, a difference below this comes from rounding
N = 400
column, by_region, subscribers, bucket = [], {}, {}, {}
for p in chunks(PATH, N):
    for s in p:
        o = float(s[4])
        column.append(o)
        by_region[s[3]] = by_region.get(s[3], 0.0) + o
        subscribers[s[0]] = o
        d = round(o * 10)                 # readings have one decimal: 321 buckets are enough
        bucket[d] = bucket.get(d, 0) + 1
running, median_hist = 0, None
for d in sorted(bucket):
    running += bucket[d]
    if median_hist is None and running >= (len(column) + 1) // 2:
        median_hist = d / 10

footprint = lambda x: sys.getsizeof(x) + sum(sys.getsizeof(a) for a in x)
G2 = (36, 12, 12, 8, 12)
fmt2 = lambda h: "".join(str(v).rjust(G2[j]) if j else str(v).ljust(G2[0])
                         for j, v in enumerate(h))
print(fmt2(["operation", "state items", "state bytes", "passes", "result"]))
for label, state, state_bytes, result in [
        ("total, count, max", 3, 3 * 24, round(sum(column), 1)),
        ("weighted mean", 2, 2 * 24, round(sum(column) / len(column), 4)),
        ("total per region", len(by_region), footprint(by_region), len(by_region)),
        ("exact median (histogram)", len(bucket), footprint(bucket), median_hist),
        ("exact median (holding the column)", len(column), footprint(column),
         sorted(column)[(len(column) - 1) // 2]),
        ("distinct subscriber count (exact)", len(subscribers), footprint(subscribers),
         len(subscribers))]:
    print(fmt2([label, state, state_bytes, 1, result]))
chunk_bytes = max(nbytes(p) for p in chunks(PATH, 100))
print(f"at chunk size 100 the chunk holds {chunk_bytes} bytes, the subscriber state "
      f"{footprint(subscribers)} bytes; the actual peak is {chunk_bytes + footprint(subscribers)} "
      f"bytes and the state share does not shrink when the chunk size changes")

operation                            state items state bytes  passes      result
total, count, max                              3          72       1     66667.9
weighted mean                                  2          48       1     20.8402
total per region                               5         413       1           5
exact median (histogram)                     321       18292       1        20.7
exact median (holding the column)           3199      102816       1        20.7
distinct subscriber count (exact)           1329       87166       1        1329
at chunk size 100 the chunk holds 36132 bytes, the subscriber state 87166 bytes; the actual peak is 123298 bytes and the state share does not shrink when the chunk size changes

The table divides the operations into three groups. Those with constant state are the easiest: the total, the count and the maximum finish with three accumulators, the weighted mean with two. For these, chunking is free. Those with bounded state come next: the total per region carries five items, the exact median 321 items. That the median can be computed chunk by chunk is not a general truth but the consequence of an assumption specific to this dataset — because readings have one decimal and lie between 5.0 and 37.0, the value space fits into 321 buckets and the histogram gives the same number without holding the column: 20.7 m³. If the values were continuous, this row would drop out of the table.

The third group is the one that eats the gain of chunking. The exact distinct subscriber count carries a state of 1,329 items, and that state is 87,166 bytes. When the chunk is reduced to 100 rows, the chunk itself takes 36,132 bytes, but the actual peak is 123,298 bytes, and the state share does not shrink as the chunk shrinks. The gain of chunking reverses here: three quarters of the peak now belongs to the operation itself. The same holds for the median computation that holds the whole column — 102,816 bytes, no different from holding the whole dataset.

The Order Assumption Chunking Makes

# order.py — the same dataset is written to a second file in subscriber order and the same
# computation is repeated in two layouts. Chunked processing silently makes an order
# assumption; when the assumption breaks no error is raised, the result empties.
PATH2 = os.path.join(DIR, "readings_subscriber.csv")
with open(PATH2, "w", newline="") as f:
    w = csv.writer(f, lineterminator="\n")
    w.writerow(HEADER)
    w.writerows(sorted(r0, key=lambda x: (x[0], x[2])))


def streaming_increase(path: str, n: int) -> dict:   # state: only the open subscriber
    increases, active, previous = {}, None, 0.0
    for p in chunks(path, n):
        for s in p:
            a, o = s[0], float(s[4])
            if a != active:
                active, previous = a, o
            else:
                increases[a] = max(increases.get(a, 0.0), o - previous)
                previous = o
    return increases


def full_increase(path: str, n: int) -> tuple[dict, int]:   # state: all subscribers
    last, increases = {}, {}
    for p in chunks(path, n):
        for s in p:
            a, o = s[0], float(s[4])
            if a in last:
                increases[a] = max(increases.get(a, 0.0), o - last[a])
            last[a] = o
    return increases, len(last)


G3 = (18, 17, 18, 14, 14)
fmt3 = lambda h: "".join(str(v).rjust(G3[j]) if j else str(v).ljust(G3[0])
                         for j, v in enumerate(h))
print(fmt3(["file layout", "streaming state", "streaming result", "full state", "full result"]))
for label, path in [("period-ordered", PATH), ("subscriber-ordered", PATH2)]:
    a1 = streaming_increase(path, N)
    a2, state = full_increase(path, N)
    print(fmt3([label, 1, len(a1), state, len(a2)]))
d1, d2 = streaming_increase(PATH2, N), full_increase(PATH, N)[0]
print(f"does the same computation give the same number in the two layouts: "
      f"{'yes' if d1 == d2 else 'no'}; largest increase "
      f"{max(d2.values()):.1f} m3")

# Moving the chunk boundary to a natural key: one chunk per region.
groups = {}
for x in r0:
    groups.setdefault(x[3], []).append(x[4])
mean_of_means = sum(sum(v) / len(v) for v in groups.values()) / len(groups)
weighted = sum(x[4] for x in r0) / len(r0)
gap = abs(mean_of_means - weighted)
print(f"chunk sizes {sorted(len(v) for v in groups.values())}; mean of the chunk means "
      f"{mean_of_means:.4f} m3, weighted mean {weighted:.4f} m3")
print(f"gap {gap:.4f} m3, {gap / THRESHOLD:.1f}x the threshold: the equal-weight assumption "
      f"breaks as soon as the chunk sizes are unequal")

file layout         streaming state  streaming result    full state   full result
period-ordered                    1                 0          1329          1035
subscriber-ordered                1              1035          1329          1035
does the same computation give the same number in the two layouts: yes; largest increase 30.7 m3
chunk sizes [319, 319, 321, 639, 1601]; mean of the chunk means 20.7363 m3, weighted mean 20.8402 m3
gap 0.1039 m3, 2.1x the threshold: the equal-weight assumption breaks as soon as the chunk sizes are unequal

The first row of the table is the hidden default of the lesson. The streaming computation carries the state of a single subscription and is therefore the cheapest option; on the subscriber-ordered file it produces an increase for 1,035 subscribers, and the result is exactly the same as the result of the computation that holds all the subscribers in memory. The same code gives a result for 0 subscribers without raising any error on the period-ordered file. An empty result does not look like a failure: the file was read, the rows were processed, the exit code is zero. Thrift in state has turned into an ordering assumption, and the assumption is not written inside the file.

The second hidden default is where the chunk is cut. When the chunk boundary is moved from a row count to a natural key — one chunk per region — the chunks range from 319 to 1,601. The mean of the chunk means gives 20.7363 m³ and the weighted mean gives 20.8402 m³; the gap is 0.1039 m³ and 2.1 times the threshold, so it cannot be counted as coming from rounding. The gap looks small, but its direction is fixed: the equal-weight assumption enlarges the small chunks, and as chunk sizes move apart from one another the deviation grows. If it is not written how a mean computed per chunk is combined, that mean counts as unmeasured.

Summary

• When the chunk size drops from 3,199 to 100, the memory peak falls from 1,152,229 bytes to 36,132 bytes, to one thirty-second; the number of passes is one and the total comes out as 66,667.9 m³ at all four chunk sizes.
• What determines whether an operation can be computed chunk by chunk is not the chunk size but the carried state: the total and the mean carry 2–3 items, the region breakdown 5 items, the exact median 321 items, the exact distinct subscriber count 1,329 items.
• That the exact median fits into 321 buckets is not a general truth but an assumption that comes from readings having one decimal and lying in the 5.0–37.0 range.
• When the chunk drops to 100 rows, the chunk itself takes 36,132 bytes but the subscriber state is 87,166 bytes; the actual peak rises to 123,298 bytes and the state share does not shrink as the chunk shrinks.
• The streaming computation gives a correct result for 1,035 subscribers on the subscriber-ordered file and, with no error, a result for 0 subscribers on the period-ordered file; chunking silently makes an order assumption.
• When the chunk boundary moves to the region, the mean of the chunk means departs from the weighted mean by 0.1039 m³, and this gap is 2.1 times the threshold.

Next Step

This lesson saved memory by reading chunk by chunk, but in every chunk it parsed all the fields of the file: for the region breakdown all five fields of the 3,199 rows read were converted from text to objects, and then three of them were thrown away. If five columns are read for the two columns the question asks about, chunking lowers only the peak; it does not lower the work done. The next lesson compares the layout that keeps the fields of a record adjacent with the layout that keeps a column adjacent in the same query, and ties the measure to a single number: how many values were read to produce the same answer.