Streaming a Big Log

Day 1’s read_lines does this:

with open(path) as file:
    return [line for line in file.read().splitlines() if line.strip()]

file.read() pulls in the whole file as one string, and splitlines() then builds a list of every line. For a seven-line sample, no problem. A busy server writes gigabytes of log a day, and a program that needs the whole file in memory before it can count anything won’t cope.

You already know the cure from the beginner course’s generators lesson: hand over one line at a time. This lesson is about applying it, measuring what it buys, and the one trap it sets.

Measure first

Don’t take the memory saving on trust. This builds a 100,000-line log in the browser’s private file system and counts its levels both ways:

import tracemalloc
from collections import Counter

# Make a big log: 100,000 lines, in this tab's private file system.
with open("big.log", "w") as file:
    for i in range(100_000):
        file.write(f"2026-09-01 09:14:02 INFO    auth: user=u{i} action=login\n")


def levels_whole_file(path):
    with open(path) as file:
        lines = file.read().splitlines()        # every line in memory at once
    return Counter(line.split(maxsplit=3)[2] for line in lines)


def read_lines(path):
    with open(path) as file:
        for line in file:                       # one line at a time
            yield line.rstrip("\n")


def levels_streamed(path):
    return Counter(line.split(maxsplit=3)[2] for line in read_lines(path))


for name, count_levels in [("whole file", levels_whole_file), ("streamed", levels_streamed)]:
    tracemalloc.start()
    result = count_levels("big.log")
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    print(f"{name:10} {dict(result)}  peak memory {peak / 1_000_000:5.1f} MB")

Same answer, a tiny fraction of the memory. And the streamed version’s peak stays flat however big the file gets, while the other grows with it.

tracemalloc reports the most memory Python’s own objects took at once while it was tracing. That’s the part of a program’s memory your code controls, and it’s the right number for comparing two ways of writing the same thing. It isn’t the whole process: here, big.log itself already sits in the browser’s in-memory file system whichever way you read it. And streaming bounds the lines, not the answers. The Counter still grows with the number of distinct keys it counts, which is fine for five levels and less fine for a million user IDs.

The analyzer as a pipeline

For the analyzer this means three stages, each a generator except the last:

lines   = read_lines("server.log")      # yields str, one at a time
records = parse_lines(lines, errors)    # yields LogRecord, one at a time
levels, error_sources = summarize(records)   # reads each record once

Nothing is read until summarize asks for the next record. Then parse_lines asks for the next line, and read_lines reads it from the file. At any moment only one line is in flight.

Bug hunt: the report that went quiet

Someone rewrote the report with the new generator pipeline. The level counts are right. The errors-by-source count is empty, and nothing raised:

from collections import Counter
from dataclasses import dataclass


@dataclass(frozen=True)
class LogRecord:
    level: str
    source: str


LINES = [
    "2026-09-01 09:14:02 INFO    auth: user=ada",
    "2026-09-01 09:15:30 ERROR   payments: order=1043",
    "2026-09-01 09:16:11 INFO    auth: user=alan",
    "2026-09-01 09:17:45 ERROR   search: query=dune",
]


def parse_lines(lines):
    for line in lines:
        date, time, level, rest = line.split(maxsplit=3)
        yield LogRecord(level, rest.partition(":")[0])


records = parse_lines(LINES)

levels = Counter(record.level for record in records)
error_sources = Counter(record.source for record in records if record.level == "ERROR")

print("levels:          ", dict(levels))
print("errors by source:", dict(error_sources))

Reproduce. Four lines, two of them errors. Small enough to count by eye: errors by source should be {'payments': 1, 'search': 1}.

Inspect. The output is wrong but nothing failed, so inspect the state between the two counts. Add print(list(records)) after the levels line. It prints [].

Narrow. records is a generator, and a generator can only be read once. The first Counter read it to the end, so the second one got nothing and quietly counted zero errors. The bug isn’t in either count. It’s in asking two questions of something that can only be read through once.

Pin it. Test the summary with a generator as its input, not a list. A list can be read as often as you like, so a test that passes a list would never catch this. That’s why the day’s test suite has test_summary_works_on_a_generator.

Fix. There are two, with different costs:

• records = list(parse_lines(LINES)) gives you something you can read twice, but it holds every record in memory again, which is exactly what streaming was meant to avoid.
• Answer every question in one pass, which is what the analyzer’s summarize does:

def summarize(records):
    """Count levels and errors per source in a single pass."""
    levels, error_sources = Counter(), Counter()
    for record in records:
        levels[record.level] += 1
        if record.level == "ERROR":
            error_sources[record.source] += 1
    return levels, error_sources

Your turn

import sys
import unittest
from collections import Counter
from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class LogRecord:
    timestamp: datetime
    level: str


def summarize_by_hour(records):
    """Return (records per hour, errors per hour) - reading records only once."""
    raise NotImplementedError("your code here")


def at(hour, level):
    return LogRecord(datetime(2026, 9, 1, hour, 30), level)


class TestSummarizeByHour(unittest.TestCase):
    def test_counts_per_hour(self):
        records = [at(9, "INFO"), at(9, "ERROR"), at(10, "ERROR"), at(10, "ERROR")]
        per_hour, errors_per_hour = summarize_by_hour(records)
        self.assertEqual(per_hour, {9: 2, 10: 2})
        self.assertEqual(errors_per_hour, {9: 1, 10: 2})

    def test_works_on_a_generator(self):
        records = (at(h, "ERROR") for h in (9, 9, 11))      # can be read only once
        per_hour, errors_per_hour = summarize_by_hour(records)
        self.assertEqual(per_hour, {9: 2, 11: 1})
        self.assertEqual(errors_per_hour, {9: 2, 11: 1})


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestSummarizeByHour)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

def summarize_by_hour(records):
    """Return (records per hour, errors per hour) - reading records only once."""
    per_hour, errors_per_hour = Counter(), Counter()
    for record in records:
        hour = record.timestamp.hour
        per_hour[hour] += 1
        if record.level == "ERROR":
            errors_per_hour[hour] += 1
    return per_hour, errors_per_hour

What you can now do

• Read a file lazily with a generator, and measure the difference with tracemalloc instead of assuming it.
• Build a pipeline of generators where only one item is in flight at a time.
• Spot, test for, and avoid the second pass over a stream.

Next: the lines that aren’t valid log lines at all.