Generators that transform streams

Lazy pipelines

Lesson 19 of 20 · 11 min · Module 6: Bigger programs

Big data does not fit in memory all at once. Generators process items one at a time, so a pipeline can handle a file of millions of lines, or an endless stream.

Start this lesson →

Free. No installation: you write and run the code in your browser, and it is checked automatically.

By the end, you’ll be able to…

Write generators that consume other iterables
Process endless streams safely
Use deque for sliding windows

The idea

A function containing yield returns a generator. Generators can take other iterables as input and yield transformed items, forming a pipeline: each step pulls the next item only when needed. Errors in a generator appear when it is iterated, not when it is created.

def evens(numbers):
    for n in numbers:
        if n % 2 == 0:
            yield n

print(list(evens(range(10))))

01Pull, do not push

In list(chunked(numbers, 3)), list asks chunked for items, and chunked asks numbers only when it needs more. Nothing is computed ahead of time, so memory use stays small.

02Errors appear on first use

Calling chunked(data, 0) only creates the generator. The ValueError is raised when the first item is requested, which is why the check above calls next().

def gen():
    raise ValueError("bad")
    yield 1

g = gen()
print("created")

Output

created

03deque(maxlen=…) slides

A deque with maxlen drops the oldest item automatically when a new one is added. It is the natural structure for "the last N values", such as moving averages in monitoring.

Your turn

Write a generator chunked(iterable, size) that yields lists of size items (the last one may be shorter) and works with endless iterators. Raise ValueError for a size below 1. Then write moving_total(numbers, window), a generator yielding the sum of the last window numbers for each position once enough numbers have arrived.

What the checks look for

Splits into chunks with a shorter last chunk
Works lazily with an endless iterator
Rejects a size below 1
Yields moving totals once the window is full

Try it in the editor →

Take this with you

Generators chain into pipelines that handle any amount of data.

Go a little further

Write a pipeline read_lines → parse_line → only ERROR lines → chunked(…, 100) that processes a log lazily.