Count errors in a 10 GB log

New to some of the words above? Open a quick recap. Each takes about two minutes.

Quick recapComprehensions in two minutes· 2 min

A comprehension builds a list from a loop in one line. An if at the end keeps only some items.

nums = [3, -1, 4]
[n * 2 for n in nums]        # [6, -2, 8]
[n for n in nums if n > 0]   # [3, 4]
sum(n for n in nums)         # 6: sum() takes the loop directly

The full lesson: I removed the bad readings, but one is still there.

Quick recapOpening a file· 2 min

open() gives you a file object. The with block closes the file for you when it ends, even after an error.

with open("notes.txt", encoding="utf-8") as f:
    text = f.read()   # the whole file, as one string
lines = text.splitlines()   # a list, one string per line

See the problem

Your app writes one line to a log file for every event. Last night, checkout broke for an hour. Your lead asks: how many errors did we log?

Here are the first five lines:

09:00:01 INFO user 17 logged in
09:00:02 ERROR payment timed out
09:00:04 WARN slow query

09:00:07 ERROR payment timed out

Counting these is easy: 2. The hard part is the size. The whole file is 10 GB, and your laptop has 8 GB of memory.

Define the task

• Input: the path to a text file. Each line is time level message, separated by spaces, like 09:00:02 ERROR payment timed out.
• Output: how many lines have the level ERROR.
• Rules: the file can be bigger than your memory. Skip blank or cut-off lines (fewer than two words). Only the level counts: 09:00:09 INFO retry after ERROR is not an error.

Example: the five lines above give 2.

Pause and think

⏸ Pause and think

Imagine the log printed on paper, far too many pages to spread out. How would you count the errors by hand?

Give it two minutes with pen and paper before you open a hint.

Hints open one at a time:

Hint 1 of 3

Do you need every page on the desk at once?

Hint 2 of 3

Read one line, decide, add to a count, move on. What do the counts need to remember?

Hint 3 of 3

In Python, looping over a file object (for line in f) gives you one line at a time.

Understand the idea

Think of a conveyor belt in a sorting room. One worker takes parcels off the truck. The next reads each label. The next keeps only the parcels marked “fragile”. The last one counts them. Each parcel moves on before the next one arrives, so the room never fills up, however big the truck is.

Python’s conveyor belt is the generator. It’s a function that uses yield instead of return. Each yield hands out one value and pauses the function until someone asks for the next one.

def countdown(n):
    while n > 0:
        yield n
        n -= 1

numbers = countdown(3)  # nothing runs yet
next(numbers)           # 3: runs up to the first yield
next(numbers)           # 2
list(numbers)           # [1]: whatever is left
list(numbers)           # []: used up

Two things to remember. A generator does nothing until asked (it’s lazy). And it runs once: after its last value, it’s used up.

★Key insight

Handle one line at a time and let it go. Then memory depends on the size of one line, not the size of the file. Generators let you split that work into small steps that still pass along one line at a time.

Watch a pipeline of four generators count the errors. The player stops to ask what happens next. Answer, or press Just show me. You can also type your own log lines below the player.

A pipeline of generators

⤢ Full view

app.log

1. 09:00:01 INFO user 17 logged in (not read yet)
2. 09:00:02 ERROR payment timed out (not read yet)
3. 09:00:04 WARN slow query (not read yet)
4. (blank line) (not read yet)
5. 09:00:07 ERROR payment timed out (not read yet)

1. read_lineswaiting
2. parsewaiting
3. only_errorswaiting
4. sumwaiting

In memory now

Nothing yet

count = 0

Step 1 of 19. sum asks only_errors for an error. only_errors asks parse, and parse asks read_lines. Nothing has been read yet.

◀ BackPlayNext ▶Reset Ask me first Show as text

1from collections.abc import Iterable, Iterator
2 
3Entry = tuple[str, str, str]  # (time, level, message)
4 
6def read_lines(path: str) -> Iterator[str]:
7    with open(path, encoding="utf-8", errors="replace") as f:
8        yield from f
9 
11def parse(lines: Iterable[str]) -> Iterator[Entry]:
12    for line in lines:
13        parts = line.split(maxsplit=2)
14        if len(parts) < 2:
15            continue  # a blank or cut-off line
16        message = parts[2].rstrip() if len(parts) == 3 else ""
17        yield parts[0], parts[1], message
18 
20def only_errors(entries: Iterable[Entry]) -> Iterator[Entry]:
21    for entry in entries:
22        if entry[1] == "ERROR":
23            yield entry
24 
26def count_errors(path: str) -> int:
27    return sum(1 for _ in only_errors(parse(read_lines(path))))

Try your own input

Log lines, one per row09:00:01 INFO user 17 logged in 09:00:02 ERROR payment timed out 09:00:04 WARN slow query 09:00:07 ERROR payment timed outShow it

Climb the ladder

All three rungs read every line once, so all three take O(n) time, where n is the size of the file. What changes is memory, and how easily the code grows.

Okay

Read it all, then loop

The idea: Read the whole file into memory, split it into lines, then count.

On the example: f.read() returns all five lines as one string. splitlines() turns it into a list of five strings. The loop checks each one: lines 2 and 5 are errors, and line 4 is blank, so it’s skipped. Count: 2.

Reading it all first

⤢ Full view

app.log

1. 09:00:01 INFO user 17 logged in (not read yet)
2. 09:00:02 ERROR payment timed out (not read yet)
3. 09:00:04 WARN slow query (not read yet)
4. (blank line) (not read yet)
5. 09:00:07 ERROR payment timed out (not read yet)

In memory now

Nothing yet

Step 1 of 10. Open the file. Nothing has been read yet.

◀ BackPlayNext ▶Reset Ask me first Show as text

1def count_errors(path: str) -> int:
2    with open(path, encoding="utf-8", errors="replace") as f:
3        lines = f.read().splitlines()  # the whole file, in memory at once
4    count = 0
5    for line in lines:
6        parts = line.split(maxsplit=2)
7        if len(parts) >= 2 and parts[1] == "ERROR":
8            count += 1
9    return count

def count_errors(path: str) -> int:
    with open(path, encoding="utf-8", errors="replace") as f:
        lines = f.read().splitlines()  # the whole file, in memory at once
    count = 0
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            count += 1
    return count

Time

O(n)

Space

O(n)

Why

f.read() holds the whole text, and splitlines() makes a list with a copy of every line: about 2.5 times the file at the peak.

Where it breaks: memory. A 10 GB file needs about 25 GB before the first line is checked. On an 8 GB laptop, Python stops with a MemoryError, or the system stops it. Even when it fits, nothing happens until the whole file is read.

Better

Stream it, line by line

The idea: Loop over the file object itself. It hands you one line at a time.

The waste in Okay: it keeps every line, but each line is needed for only a moment.

On the example: Python reads line 1, the loop checks it, and it’s gone before line 2 is read. Line 4 is blank, so parts has fewer than two items and it’s skipped. Count: 2.

Highlighted lines are new in this rung.

def count_errors(path: str) -> int:
    count = 0
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:  # one line at a time
            parts = line.split(maxsplit=2)
            if len(parts) >= 2 and parts[1] == "ERROR":
                count += 1
    return count

Time

O(n)

Space

O(1)was O(n)

Why

Only the current line and a small read buffer are in memory, whatever the file size. (Strictly, memory grows with the longest line.)

What changed from the previous rung

def count_errors(path: str) -> int:
    with open(path, encoding="utf-8", errors="replace") as f:
        lines = f.read().splitlines()  # the whole file, in memory at once
    count = 0
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            count += 1
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:  # one line at a time
            parts = line.split(maxsplit=2)
            if len(parts) >= 2 and parts[1] == "ERROR":
                count += 1
    return count

Where it breaks: one loop does every job: reading, splitting, filtering and counting. The next question (which error happens most? only the last hour? yesterday’s zipped logs too?) means copying the loop and changing the middle. And you can’t test the splitting without a real file.

Best

A pipeline of generators

The idea: Give each job its own small generator, chain them, and count what comes out of the end.

The waste in Better: every job is mixed into one loop, so nothing can be reused or tested alone.

On the example: sum asks for an error, and the request travels back to read_lines, which reads line 1. parse splits it, and only_errors drops it (it’s INFO) and asks for another. Line 2 is an ERROR, so it reaches sum: 1. Line 4 is blank, and parse skips it. Line 5 makes 2.

yield from f hands out each line of f, one at a time. Iterator[str] is a type hint: “gives strings, one at a time”.

Highlighted lines are new in this rung.

from collections.abc import Iterable, Iterator

Entry = tuple[str, str, str]  # (time, level, message)


def read_lines(path: str) -> Iterator[str]:
    with open(path, encoding="utf-8", errors="replace") as f:
        yield from f


def parse(lines: Iterable[str]) -> Iterator[Entry]:
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) < 2:
            continue  # a blank or cut-off line
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def only_errors(entries: Iterable[Entry]) -> Iterator[Entry]:
    for entry in entries:
        if entry[1] == "ERROR":
            yield entry


def count_errors(path: str) -> int:
    return sum(1 for _ in only_errors(parse(read_lines(path))))

Time

O(n)

Space

O(1)

Why

Each stage holds one item at a time, so memory stays as small as in Better.

What changed from the previous rung

def count_errors(path: str) -> int:
    count = 0
from collections.abc import Iterable, Iterator

Entry = tuple[str, str, str]  # (time, level, message)


def read_lines(path: str) -> Iterator[str]:
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:  # one line at a time
            parts = line.split(maxsplit=2)
            if len(parts) >= 2 and parts[1] == "ERROR":
                count += 1
    return count
        yield from f


def parse(lines: Iterable[str]) -> Iterator[Entry]:
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) < 2:
            continue  # a blank or cut-off line
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def only_errors(entries: Iterable[Entry]) -> Iterator[Entry]:
    for entry in entries:
        if entry[1] == "ERROR":
            yield entry


def count_errors(path: str) -> int:
    return sum(1 for _ in only_errors(parse(read_lines(path))))

When Best isn’t best: for a one-off count on Linux or macOS, a shell one-liner is quicker to type and faster to run: awk '$2 == "ERROR" { n++ } END { print n + 0 }' app.log. And if parsing each line is slow, one CPU core becomes the limit. See below.

Why the pipeline is worth it

New questions become a new stage, not a new loop. Each function below reuses the Best rung’s stages:

Growing the pipeline

import gzip
from collections import Counter
from collections.abc import Iterable, Iterator


def top_messages(path: str, k: int) -> list[tuple[str, int]]:
    """The k most common error messages: no new loop, just a new last stage."""
    errors = only_errors(parse(read_lines(path)))
    return Counter(message for _, _, message in errors).most_common(k)


def read_gzip_lines(path: str) -> Iterator[str]:
    """A new first stage for gzipped logs. The rest of the pipeline stays the same."""
    with gzip.open(path, "rt", encoding="utf-8", errors="replace") as f:
        yield from f


def read_many(paths: Iterable[str]) -> Iterator[str]:
    """Several files, one after another, still one line at a time."""
    for path in paths:
        yield from read_lines(path)

Each stage also works on a plain list, so you can test parse with three strings and no file at all.

When one core isn’t enough

Reading a file is usually limited by the disk. But if each line needs heavy work, like parsing JSON, Python’s one core can become the limit. Then split the file into chunks by byte position and count each chunk in its own process.

Count in chunks, on several cores

import os
from concurrent.futures import ProcessPoolExecutor


def count_in_chunk(path: str, start: int, end: int) -> int:
    """Count the ERROR lines that begin at a byte from start up to (not including) end."""
    count = 0
    with open(path, "rb") as f:
        if start > 0:
            f.seek(start - 1)
            f.readline()  # skip to the next line start; the chunk before owns this line
        while f.tell() < end:
            line = f.readline()
            if not line:
                break
            parts = line.split(maxsplit=2)
            if len(parts) >= 2 and parts[1] == b"ERROR":
                count += 1
    return count


def count_errors_parallel(path: str, workers: int = 4) -> int:
    size = os.path.getsize(path)
    step = size // workers + 1
    starts = range(0, size, step)
    with ProcessPoolExecutor(workers) as pool:
        jobs = [pool.submit(count_in_chunk, path, s, s + step) for s in starts]
        return sum(job.result() for job in jobs)

• Each chunk owns the lines that start inside it. A chunk that begins mid-line skips to the next line start. The chunk before finishes that line.
• Bytes, not text: a byte position can land in the middle of a character, so each chunk reads bytes and compares with b"ERROR".
• Measure first. If the disk is the limit, more processes won’t help.

This version starts new processes, so it can’t run in the browser. It’s tested on Linux and Windows.

Write it yourself

Now write it in the page. Press Run tests to check your code. Python runs in your browser, so nothing is sent anywhere. Your code is saved in this browser for next time.

Try the Okay version first: it passes the small tests but fails the memory test. Then make it stream.

Count errors in a log

Ctrl + Enter runs

def count_errors(path: str) -> int: # Count the lines whose level # (the second word) is ERROR. # The file may not fit in memory. return 0

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: count_errors('app.log') → 2
   What’s in app.log

   09:00:01 INFO user 17 logged in
   09:00:02 ERROR payment timed out
   09:00:04 WARN slow query
   
   09:00:07 ERROR payment timed out
   

2. ○Not run yet: count_errors('retry.log') → 1
   What’s in retry.log

   09:00:01 INFO retry after ERROR
   09:00:02 ERROR disk full
   

3. ○Not run yet: Extra test
4. ○Not run yet: Extra test
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

count_errors()Run onceVisualize

Compare

Rung	Time	Memory	Good	Not so good	Use it when
Okay	O(n)	O(n), about 2.5 × the file	Short; every line stays at hand	Runs out of memory on big files	The file is small for sure
Better	O(n)	O(1)	Tiny memory, one simple loop	One loop does every job	A quick one-off script
Best	O(n)	O(1)	Tiny memory; stages you can test, swap, reuse	More functions to read	Code that will grow or run often

How much memory is that? Each bar below is one rung, for a file of the size you pick. The line is the computer’s memory. Press Read the file to watch it fill.

Memory needed to count the errors

1. OkayRead it all, then loop≈ 25 GB: doesn't fit in 8 GB

2. BetterStream it, line by line≈ 150 KB: fits

3. BestA pipeline of generators≈ 150 KB: fits

The line is this computer's memory: 8 GB.

Estimates for lines of about 80 characters. Reading it all holds the text and a list of every line, about 2.5 times the file. Streaming holds one line and a small buffer.

Common mistakes and edge cases

!A generator runs once

After a generator gives its last value, it’s used up. A second loop over it sees nothing, with no error.

errors = only_errors(parse(read_lines("app.log")))
total = sum(1 for _ in errors)               # reads every error
top = Counter(e[2] for e in errors)          # empty: errors is used up

Build a new pipeline for each question, or answer both in one pass.

• "ERROR" in line is wrong: it also counts INFO retry after ERROR. Check the level, the second word.
• One bad byte: a single invalid byte in 10 GB stops the read with a UnicodeDecodeError, maybe hours in. errors="replace" swaps it for � and keeps going.
• readlines() isn’t streaming: it builds a list of every line, just like read().splitlines().
• len() doesn’t work on a generator. Count with sum(1 for _ in gen).
• Stopping early: a file opened inside a generator stays open until the generator finishes or is closed. In a long-running program, call .close() on a generator you stop using.
• Windows line endings: text mode turns \r\n into \n for you.

Say it in the interview

Questions to ask first

• How big is the file, and how much memory do we have?
• One file or many? Are some compressed?
• Is the format fixed? Can lines be broken?
• Is this a one-off answer, or a job that runs every hour?

A 60-second walkthrough

“Reading the whole file would need more memory than we have, so I’ll stream it. A Python file object gives one line at a time, so memory stays flat. I’ll split the work into a pipeline of generators: read lines, parse each into time, level and message, keep the errors, and count. Each stage pulls one item at a time, so it’s O(n) time and O(1) memory. And I can test each stage with a plain list. I’ll check the level field, not ‘ERROR’ anywhere in the line, and skip broken lines. If parsing turned out to be the slow part, I’d split the file into byte ranges and count them on several cores.”

Likely follow-ups

• The top 5 error messages? Feed the errors into Counter and call .most_common(5). Memory grows with the number of different messages, not lines.
• The file is still being written? Keep reading at the end and wait for new lines, like tail -f. A generator can yield forever.
• Logs on 100 servers? Count on each server, then add up the counts. Only the small counts travel.
• Make it faster? Measure first. If the disk is the limit, read bigger blocks. If parsing is, use chunks on several cores.
• Why not readlines()? It builds a list of every line: the same memory problem as read().

Red flags interviewers notice

• Calling read() or readlines() on a file of unknown size.
• Saying generators make code faster. They save memory; the time is about the same.
• Counting "ERROR" in line.
• Looping over the same generator twice.

Practice

Four related problems, from easy to hard. In each one, lines can be a list of strings, an open file or a generator. Your code shouldn’t care which. Each has hints, tests you can run here, and its own Okay, Better and Best answers. Try before you open them.

1. Top error messages (easy)

Return the k most common error messages with their counts, most common first. If two messages tie, the one seen first comes first.

Example: on the log in the first test, top_errors(lines, 2) gives [("payment timed out", 3), ("disk full", 2)].

Hint 1 of 3

Start from a stage that gives only the messages of ERROR lines.

Hint 2 of 3

Counting how often each value appears is a dict job: message → count.

Hint 3 of 3

collections.Counter counts for you, and .most_common(k) returns the top k in order, ties as first seen.

Top error messages

Ctrl + Enter runs

def top_errors(lines, k: int) -> list[tuple[str, int]]: # Return the k most common ERROR messages as (message, count), most common first. return []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: top_errors(['09:00:01 ERROR payment timed out', '09:00:02 INFO cart saved', '09:00:03 ERROR disk full', '09:00:04 ERROR payment timed out', '', '09:00:05 WARN slow query', '09:00:06 ERROR login failed', '09:00:07 ERROR disk full', '09:00:08 ERROR payment timed out'], 2) → [['payment timed out', 3], ['disk full', 2]]
2. ○Not run yet: top_errors(['09:00:01 ERROR b', '09:00:02 ERROR a', '09:00:03 ERROR a', '09:00:04 ERROR b'], 1) → [['b', 2]]
3. ○Not run yet: Extra test
4. ○Not run yet: Extra test
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

top_errors()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Count each message with list.count

The idea: Collect every error message, then count each different one by scanning the list.

Fine for a few hundred lines. With thousands of different messages, the scans add up fast.

from collections.abc import Iterable


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    messages = []
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            messages.append(parts[2].rstrip() if len(parts) == 3 else "")
    unique = []
    for message in messages:
        if message not in unique:
            unique.append(message)
    counts = [(message, messages.count(message)) for message in unique]
    counts.sort(key=lambda pair: pair[1], reverse=True)
    return counts[:k]

Time

O(n · u)

Space

O(n)

Why

u different messages, and each count scans all n errors. The list keeps every error.

Better

Count in a dict, then sort

The idea: One pass adds 1 to each message's count in a dict; then sort the counts.

Memory now grows with the number of different messages, not the number of lines.

Highlighted lines are new in this rung.

from collections.abc import Iterable


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    counts: dict[str, int] = {}
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            message = parts[2].rstrip() if len(parts) == 3 else ""
            counts[message] = counts.get(message, 0) + 1
    ranked = sorted(counts.items(), key=lambda pair: pair[1], reverse=True)
    return ranked[:k]

Time

O(n + u log u)was O(n · u)

Space

O(u)was O(n)

Why

One pass over the lines, then a sort of the u different messages. Python's sort keeps ties in the order first seen.

What changed from the previous rung

from collections.abc import Iterable


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    messages = []
    counts: dict[str, int] = {}
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            messages.append(parts[2].rstrip() if len(parts) == 3 else "")
    unique = []
    for message in messages:
        if message not in unique:
            unique.append(message)
    counts = [(message, messages.count(message)) for message in unique]
    counts.sort(key=lambda pair: pair[1], reverse=True)
    return counts[:k]
            message = parts[2].rstrip() if len(parts) == 3 else ""
            counts[message] = counts.get(message, 0) + 1
    ranked = sorted(counts.items(), key=lambda pair: pair[1], reverse=True)
    return ranked[:k]

Best

Counter on the error stage

The idea: A generator of error messages feeds collections.Counter, and most_common(k) picks the top k.

Same idea as Better, with the counting and ranking done by the standard library.

Highlighted lines are new in this rung.

from collections import Counter
from collections.abc import Iterable, Iterator


def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    messages = (message for _, level, message in parse(lines) if level == "ERROR")
    return Counter(messages).most_common(k)

Time

O(n + u log k)was O(n + u log u)

Space

O(u)

Why

Counter does the same one pass. most_common(k) keeps only the k best while it looks, instead of sorting everything.

What changed from the previous rung

from collections.abc import Iterable
from collections import Counter
from collections.abc import Iterable, Iterator


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    counts: dict[str, int] = {}
def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            message = parts[2].rstrip() if len(parts) == 3 else ""
            counts[message] = counts.get(message, 0) + 1
    ranked = sorted(counts.items(), key=lambda pair: pair[1], reverse=True)
    return ranked[:k]
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def top_errors(lines: Iterable[str], k: int) -> list[tuple[str, int]]:
    messages = (message for _, level, message in parse(lines) if level == "ERROR")
    return Counter(messages).most_common(k)

2. First errors from an endless log (easy)

Return the messages of the first n errors, in order, or fewer if the log ends first. The log may never end: a live log keeps growing. So stop reading as soon as you have n.

Example: on the log in the first test, first_errors(lines, 2) gives ["one", "two"].

Hint 1 of 3

What happens to a loop over a log that never ends, if you wait for the end?

Hint 2 of 3

Once you have n errors, you can stop: break out of the loop.

Hint 3 of 3

itertools.islice(items, n) takes the first n items and stops asking for more.

First errors from an endless log

Ctrl + Enter runs

def first_errors(lines, n: int) -> list[str]: # Return the messages of the first n ERROR lines. The log may never end. return []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: first_errors(['09:00:01 INFO start', '09:00:02 ERROR one', '', '09:00:03 ERROR two', '09:00:04 WARN slow', '09:00:05 ERROR three'], 2) → ['one', 'two']
2. ○Not run yet: first_errors(['09:00:01 INFO start', '09:00:02 ERROR one', '', '09:00:03 ERROR two', '09:00:04 WARN slow', '09:00:05 ERROR three'], 10) → ['one', 'two', 'three']
3. ○Not run yet: Extra test
4. ○Not run yet: Extra test
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

first_errors()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Collect every error, then slice

The idea: Gather all the errors in a list, then return the first n.

Works on a finished file, but does far more work than needed, and hangs on a live log.

from collections.abc import Iterable


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors = []
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            errors.append(parts[2].rstrip() if len(parts) == 3 else "")
    return errors[:n]

Time

O(whole log)

Space

O(errors)

Why

It reads to the end before slicing. On a log that never ends, it never returns.

Better

Stop with break

The idea: Append each error and break out of the loop once there are n.

The n <= 0 check matters: without it, the loop would never hit len(errors) == n and would read forever.

Highlighted lines are new in this rung.

from collections.abc import Iterable


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors: list[str] = []
    if n <= 0:
        return errors
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            errors.append(parts[2].rstrip() if len(parts) == 3 else "")
            if len(errors) == n:
                break  # we have enough: stop reading
    return errors

Time

O(lines up to the nth error)was O(whole log)

Space

O(n)was O(errors)

Why

Reading stops at the nth error. Only n messages are kept.

What changed from the previous rung

from collections.abc import Iterable


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors = []
    errors: list[str] = []
    if n <= 0:
        return errors
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            errors.append(parts[2].rstrip() if len(parts) == 3 else "")
    return errors[:n]
            if len(errors) == n:
                break  # we have enough: stop reading
    return errors

Best

islice on the error stage

The idea: Make a generator of error messages and take the first n with itertools.islice.

The stopping rule lives in one place, islice, so n = 0 needs no special case.

Highlighted lines are new in this rung.

from collections.abc import Iterable, Iterator
from itertools import islice


def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors = (message for _, level, message in parse(lines) if level == "ERROR")
    return list(islice(errors, n))

Time

O(lines up to the nth error)

Space

O(n)

Why

islice asks for exactly n items, and the lazy pipeline reads only the lines needed to produce them.

What changed from the previous rung

from collections.abc import Iterable
from collections.abc import Iterable, Iterator
from itertools import islice


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors: list[str] = []
    if n <= 0:
        return errors
def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            errors.append(parts[2].rstrip() if len(parts) == 3 else "")
            if len(errors) == n:
                break  # we have enough: stop reading
    return errors
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def first_errors(lines: Iterable[str], n: int) -> list[str]:
    errors = (message for _, level, message in parse(lines) if level == "ERROR")
    return list(islice(errors, n))

3. Error spikes per minute (medium)

Return the minutes ("HH:MM") that have more than limit errors, in time order. The log is in time order.

Example: on the log in the first test, error_spikes(lines, 2) gives ["09:14", "09:16"].

Hint 1 of 3

The minute is the first five characters of the time: "09:14:05"[:5] is "09:14".

Hint 2 of 3

A dict from minute to error count answers the question in one pass.

Hint 3 of 3

The log is in time order, so all lines of one minute sit together. itertools.groupby can hand you one minute at a time.

Error spikes per minute

Ctrl + Enter runs

def error_spikes(lines, limit: int) -> list[str]: # Return each minute ("HH:MM") with more than limit ERROR lines, in time order. return []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: error_spikes(['09:14:01 ERROR payment timed out', '09:14:20 ERROR payment timed out', '09:14:59 ERROR disk full', '09:15:02 INFO recovered', '09:15:30 ERROR payment timed out', '', '09:16:00 ERROR a', '09:16:01 ERROR b', '09:16:02 ERROR c'], 2) → ['09:14', '09:16']
2. ○Not run yet: error_spikes(['09:14:01 ERROR payment timed out', '09:14:20 ERROR payment timed out', '09:14:59 ERROR disk full', '09:15:02 INFO recovered', '09:15:30 ERROR payment timed out', '', '09:16:00 ERROR a', '09:16:01 ERROR b', '09:16:02 ERROR c'], 0) → ['09:14', '09:15', '09:16']
3. ○Not run yet: Extra test
4. ○Not run yet: Extra test
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

error_spikes()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Count each minute with list.count

The idea: List the minute of every error, then count each minute by scanning the list.

Fine for a short log; an afternoon of errors makes it very slow.

from collections.abc import Iterable


def error_spikes(lines: Iterable[str], limit: int) -> list[str]:
    minutes = []
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            minutes.append(parts[0][:5])  # "09:14:05" -> "09:14"
    spikes = []
    for minute in minutes:
        if minute not in spikes and minutes.count(minute) > limit:
            spikes.append(minute)
    return spikes

Time

O(e²)

Space

O(e)

Why

For e errors, each one may trigger a scan of all e. The list keeps every error's minute.

Better

Count per minute in a dict

The idea: One pass adds 1 to each minute's count; then keep the minutes over the limit.

Fast, but it answers only at the end of the log, so it can’t warn you during a live spike.

Highlighted lines are new in this rung.

from collections.abc import Iterable


def error_spikes(lines: Iterable[str], limit: int) -> list[str]:
    counts: dict[str, int] = {}
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            minute = parts[0][:5]  # "09:14:05" -> "09:14"
            counts[minute] = counts.get(minute, 0) + 1
    return [minute for minute, count in counts.items() if count > limit]

Time

O(n)was O(e²)

Space

O(m)was O(e)

Why

One pass over n lines. The dict holds one count per minute (m minutes). Dicts keep insertion order, so the minutes stay in time order.

What changed from the previous rung

from collections.abc import Iterable


def error_spikes(lines: Iterable[str], limit: int) -> list[str]:
    minutes = []
    counts: dict[str, int] = {}
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            minutes.append(parts[0][:5])  # "09:14:05" -> "09:14"
    spikes = []
    for minute in minutes:
        if minute not in spikes and minutes.count(minute) > limit:
            spikes.append(minute)
    return spikes
            minute = parts[0][:5]  # "09:14:05" -> "09:14"
            counts[minute] = counts.get(minute, 0) + 1
    return [minute for minute, count in counts.items() if count > limit]

Best

groupby, one minute at a time

The idea: Group neighbouring lines by minute with itertools.groupby and yield each minute that spikes.

Each spike comes out as soon as the next minute’s first line arrives, so this works on a live log. It returns a generator; the tests read it into a list.

Highlighted lines are new in this rung.

from collections.abc import Iterable, Iterator
from itertools import groupby


def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def error_spikes(lines: Iterable[str], limit: int) -> Iterator[str]:
    by_minute = groupby(parse(lines), key=lambda entry: entry[0][:5])  # "09:14:05" -> "09:14"
    for minute, entries in by_minute:
        if sum(1 for _, level, _ in entries if level == "ERROR") > limit:
            yield minute  # reported as soon as the next minute's first line arrives

Time

O(n)

Space

O(1)was O(m)

Why

groupby holds one minute's group at a time, and the count is done as the group passes.

What changed from the previous rung

from collections.abc import Iterable
from collections.abc import Iterable, Iterator
from itertools import groupby


def error_spikes(lines: Iterable[str], limit: int) -> list[str]:
    counts: dict[str, int] = {}
def parse(lines: Iterable[str]) -> Iterator[tuple[str, str, str]]:
    """The parse stage from the lesson: (time, level, message), skipping broken lines."""
    for line in lines:
        parts = line.split(maxsplit=2)
        if len(parts) >= 2 and parts[1] == "ERROR":
            minute = parts[0][:5]  # "09:14:05" -> "09:14"
            counts[minute] = counts.get(minute, 0) + 1
    return [minute for minute, count in counts.items() if count > limit]
        if len(parts) < 2:
            continue
        message = parts[2].rstrip() if len(parts) == 3 else ""
        yield parts[0], parts[1], message


def error_spikes(lines: Iterable[str], limit: int) -> Iterator[str]:
    by_minute = groupby(parse(lines), key=lambda entry: entry[0][:5])  # "09:14:05" -> "09:14"
    for minute, entries in by_minute:
        if sum(1 for _, level, _ in entries if level == "ERROR") > limit:
            yield minute  # reported as soon as the next minute's first line arrives

4. The k slowest requests (hard)

Each line is a request, like 09:00:01 INFO GET /cart 182ms. Return the k slowest as (ms, path) pairs, slowest first. No two requests take the same time. Keep memory small: the log can be huge.

Example: on the log in the first test, slowest(lines, 2) gives [(960, "/pay"), (410, "/search")].

Hint 1 of 3

Sorting every request needs all of them in memory. Do you need them all?

Hint 2 of 3

Keep only the k slowest seen so far. A new request matters only if it beats the fastest of those.

Hint 3 of 3

heapq keeps the smallest item at heap[0], and adds or removes an item in O(log k).

The k slowest requests

Ctrl + Enter runs

def slowest(lines, k: int) -> list[tuple[int, str]]: # Each line looks like: 09:00:01 INFO GET /cart 182ms # Return the k slowest as (ms, path), slowest first. return []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: slowest(['09:00:01 INFO GET /cart 182ms', '09:00:02 INFO GET /home 35ms', '09:00:03 INFO POST /pay 960ms', '09:00:04 INFO GET /search 410ms', '09:00:05 INFO GET /home 12ms'], 2) → [[960, '/pay'], [410, '/search']]
2. ○Not run yet: slowest(['09:00:01 INFO GET /cart 182ms', '09:00:02 INFO GET /home 35ms', '09:00:03 INFO POST /pay 960ms', '09:00:04 INFO GET /search 410ms', '09:00:05 INFO GET /home 12ms'], 1) → [[960, '/pay']]
3. ○Not run yet: Extra test
4. ○Not run yet: Extra test
5. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

slowest()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Keep every request and sort

The idea: Put every (ms, path) in a list, sort it slowest first, and take k.

Right answer, but on a huge log the list alone can fill memory.

from collections.abc import Iterable


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    requests = []
    for line in lines:
        _, _, _, path, took = line.split()  # 09:00:01 INFO GET /cart 182ms
        requests.append((int(took.removesuffix("ms")), path))
    requests.sort(reverse=True)
    return requests[:k]

Time

O(n log n)

Space

O(n)

Why

Sorting n requests costs O(n log n), and the list holds all of them.

Better

Keep a sorted list of the k slowest

The idea: Keep only the k slowest so far, in order; insert a new one with bisect.insort and drop the fastest.

Memory is now tiny. Time is fine for small k, but grows with k.

Highlighted lines are new in this rung.

from bisect import insort
from collections.abc import Iterable


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    top: list[tuple[int, str]] = []  # the k slowest so far, fastest first
    for line in lines:
        _, _, _, path, took = line.split()  # 09:00:01 INFO GET /cart 182ms
        request = (int(took.removesuffix("ms")), path)
        if len(top) < k or request > top[0]:
            insort(top, request)
            if len(top) > k:
                top.pop(0)
    return top[::-1]

Time

O(n · k)was O(n log n)

Space

O(k)was O(n)

Why

Inserting into a sorted list of k shifts up to k items, for each of n requests in the worst case.

What changed from the previous rung

from bisect import insort
from collections.abc import Iterable


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    requests = []
    top: list[tuple[int, str]] = []  # the k slowest so far, fastest first
    for line in lines:
        _, _, _, path, took = line.split()  # 09:00:01 INFO GET /cart 182ms
        requests.append((int(took.removesuffix("ms")), path))
    requests.sort(reverse=True)
    return requests[:k]
        request = (int(took.removesuffix("ms")), path)
        if len(top) < k or request > top[0]:
            insort(top, request)
            if len(top) > k:
                top.pop(0)
    return top[::-1]

Best

A min-heap of the k slowest

The idea: Keep the k slowest in a min-heap; heap[0] is the one to beat, and heappushpop swaps in a slower request.

heapq.nlargest(k, requests(lines)) does the same in one line. Knowing how it works is what interviews check.

Highlighted lines are new in this rung.

import heapq
from collections.abc import Iterable, Iterator


def requests(lines: Iterable[str]) -> Iterator[tuple[int, str]]:
    for line in lines:
        _, _, _, path, took = line.split()  # 09:00:01 INFO GET /cart 182ms
        yield int(took.removesuffix("ms")), path


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    heap: list[tuple[int, str]] = []  # a min-heap: the fastest of the k slowest is heap[0]
    for request in requests(lines):
        if len(heap) < k:
            heapq.heappush(heap, request)
        elif request > heap[0]:
            heapq.heappushpop(heap, request)
    return sorted(heap, reverse=True)

Time

O(n log k)was O(n · k)

Space

O(k)

Why

Each push or swap on a heap of k items costs O(log k).

What changed from the previous rung

from bisect import insort
from collections.abc import Iterable
import heapq
from collections.abc import Iterable, Iterator


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    top: list[tuple[int, str]] = []  # the k slowest so far, fastest first
def requests(lines: Iterable[str]) -> Iterator[tuple[int, str]]:
    for line in lines:
        _, _, _, path, took = line.split()  # 09:00:01 INFO GET /cart 182ms
        request = (int(took.removesuffix("ms")), path)
        if len(top) < k or request > top[0]:
            insort(top, request)
            if len(top) > k:
                top.pop(0)
    return top[::-1]
        yield int(took.removesuffix("ms")), path


def slowest(lines: Iterable[str], k: int) -> list[tuple[int, str]]:
    heap: list[tuple[int, str]] = []  # a min-heap: the fastest of the k slowest is heap[0]
    for request in requests(lines):
        if len(heap) < k:
            heapq.heappush(heap, request)
        elif request > heap[0]:
            heapq.heappushpop(heap, request)
    return sorted(heap, reverse=True)

Check yourself

A few quick questions. Ones you miss are remembered, so you can find your weak spots later.

1. Predict the outputWhat does this print, in order?

   def numbers():
       print("start")
       yield 1
       print("middle")
       yield 2
   
   gen = numbers()
   print("made it")
   print(next(gen))

   start, made it, 1made it, start, 1made it, start, 1, middlestart, middle, made it, 1

   Check answer
2. Spot the bugThis code should print the number of errors, then the top 3 messages. What goes wrong?

   errors = only_errors(parse(read_lines("app.log")))
   print(sum(1 for _ in errors))
   print(Counter(e[2] for e in errors).most_common(3))

   Nothing: both lines print what they shouldThe second print shows [] because errors is used upThe first line reads the whole file into memoryCounter can't take a generator

   Check answer
3. Multiple choiceWhich of these holds every line of the file in memory at once?

   with open("app.log", encoding="utf-8") as f:
       ...

   for line in f:f.readlines()next(f)(line for line in f)

   Check answer
4. Pick the patternYou want the first 10 errors from a live log that never ends. Which fits?

   list(errors)[:10]itertools.islice(errors, 10)sorted(errors)[:10]errors[:10]

   Check answer
5. Fill in the answerFill in the blank: a function becomes a generator when it uses ____ to hand out values.

   Your answerCheck answer

Flashcards

Add these to your review deck. Each card comes back a little later every time you get it right.

• Card 1What does yield do?Show answer
• Card 2How much memory does `for line in f` use?Show answer
• Card 3Why is f.read().splitlines() risky on a big file?Show answer
• Card 4How many times can you loop over a generator?Show answer
• Card 5How do you count the items a generator gives?Show answer

Add 5 cards to my review deck

Cheat sheet

• Loop over the file object (for line in f) to read one line at a time.
• Never read() or readlines() a file of unknown size.
• One small generator per job. Chain them, then count, rank or stop at the end.
• Generators save memory, not time, and they run once.
• Check the level field, open with errors="replace", and skip broken lines.

Go deeper

• Generators, from the official Functional Programming HOWTO.
• Reading and writing files, from the Python tutorial.
• itertools: islice, groupby and more.
• collections.Counter and heapq, used in the practice answers.
• Next on the Python roadmap: collections, itertools and functools and Context managers.

Can you do these now?

If you’re not sure of any of these, scroll back to that part before moving on.

• Explain why reading a whole file can run out of memory
• Read a file one line at a time, in about the same memory for any size
• Build a pipeline of generators and say when each stage runs
• Avoid the classic slips, like looping over a generator twice

✓Mark as completeI already know this

Status: Not started