P13 Chunked processing · P14 Retry with backoff

FreeLesson 64 of 78

P13Chunked processingDEMEDIUM

Sounds like: "the file is 50 GB and memory is 8 GB", "insert in batches of 1,000", "process an S3 object stream"

Worked question: Process orders.csv four rows at a time (pretend it is 50 GB) and keep a running delivered total.

Approach: 1. stream, never read() the whole file 2. cut the stream into fixed-size batches 3. carry state between batches

SOLUTIONPYTHON 3.12

import csv
from itertools import islice

def read_in_chunks(path, size):
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        while chunk := list(islice(reader, size)):
            yield chunk                    # only `size` rows in RAM

running = 0.0
for i, chunk in enumerate(read_in_chunks("orders.csv", 4), 1):
    part = sum(float(r["amount"]) for r in chunk
               if r["status"] == "delivered")
    running += part
    print(f"chunk {i}: {len(chunk)} rows  +{part:>8,.2f}"
          f"  running {running:>9,.2f}")

Output

chunk 1: 4 rows  +3,493.00  running  3,493.00
chunk 2: 4 rows  +5,894.00  running  9,387.00
chunk 3: 4 rows  +5,590.00  running 14,977.00
chunk 4: 3 rows  +3,995.00  running 18,972.00

P14Retry with backoffDEMEDIUM

Sounds like: "the API returns 429 sometimes", "the warehouse times out at night", "make this call resilient"

Worked question: Retry a rate-limited API with exponential backoff and full jitter, and give up after a fixed number of tries.

Approach: 1. catch only retryable errors 2. wait base x 2^n, capped 3. add random jitter, stop after N

SOLUTIONPYTHON 3.12

import random, time

random.seed(7)                  # fixed seed so this page is repeatable

def backoff(base=1.0, cap=30.0, tries=6):
    for n in range(tries):
        yield round(random.uniform(0, min(cap, base * 2 ** n)), 2)
print("planned waits:", list(backoff()))

class RateLimited(Exception):
    pass

replies = iter([RateLimited("429"), RateLimited("429"), "200 OK"])
def call_api():
    r = next(replies)
    if isinstance(r, Exception):
        raise r
    return r

def with_retry(fn, tries=5, base=0.01):
    for n in range(tries):
        try:
            return fn()
        except RateLimited as e:
            wait = random.uniform(0, base * 2 ** n)
            print(f"try {n + 1}: {e}, sleep {wait:.4f}s")
            time.sleep(wait)
    raise RuntimeError("gave up after retries")

print(with_retry(call_api))

Output

planned waits: [0.32, 0.3, 2.6, 0.58, 8.57, 10.97]
try 1: 429, sleep 0.0006s
try 2: 429, sleep 0.0101s
200 OK

Practice before you move on

Solve the two practice questions for P13 and P14 in Coding Practice, under Python.
Change the chunk size in P13 to 3 and check that the running total ends at the same value.

In Coding Practice

Solve this pattern on fresh data, under Python. Hidden tests grade it.