Lazy Evaluation and Streaming Data

What is already lazy

print(range(1_000_000))                    # range(0, 1000000) - nothing built
print(map(str, [1, 2, 3]))                 # <map object>
print(filter(None, [0, 1, 2]))             # <filter object>
print(zip([1, 2], "ab"))                   # <zip object>
print(enumerate("ab"))                     # <enumerate object>
print({"a": 1}.items())                    # a view, not a list
print(open("notes.txt", encoding="utf-8")) # a file object, read on demand

All of these produce values only when iterated. Wrapping them in list() is what forces the work to happen.

import sys

print(sys.getsizeof(range(10_000_000)))          # about 48 bytes
print(sys.getsizeof(list(range(10_000_000))))    # about 80 MB

Lazy means the work may never happen

def expensive(n):
    print(f"  computing {n}")
    return n * n


values = map(expensive, [1, 2, 3])
print("map created, nothing computed yet")

print(next(iter(values)))      # only the first item is computed

numbers = range(1, 10_000_001)

# Stops at the first match; the remaining ten million are never examined
print(any(n % 9_999_991 == 0 for n in numbers))

# Stops at the first failure
print(all(n > 0 for n in numbers))

any and all short circuit. Feeding them a generator rather than a list means the values after the decision point are never produced at all.

Streaming a file

# Eager: the whole file in memory
with open("app.log", encoding="utf-8") as handle:
    lines = handle.readlines()
    errors = [line for line in lines if "ERROR" in line]
    print(len(errors))

# Lazy: one line at a time, constant memory
with open("app.log", encoding="utf-8") as handle:
    print(sum(1 for line in handle if "ERROR" in line))

The first version needs memory proportional to the file. The second needs memory proportional to the longest line. On a ten gigabyte log the difference is between impossible and instant.

Building a lazy pipeline

def read_lines(path):
    with open(path, encoding="utf-8") as handle:
        for line in handle:
            yield line.rstrip("\n")


def parse_csv(lines):
    header = next(lines).split(",")
    for line in lines:
        values = line.split(",")
        if len(values) == len(header):
            yield dict(zip(header, values))


def convert_types(records, fields):
    for record in records:
        for name, converter in fields.items():
            try:
                record[name] = converter(record[name])
            except (ValueError, KeyError):
                record[name] = None
        yield record


def where(records, predicate):
    for record in records:
        if predicate(record):
            yield record


def select(records, *names):
    for record in records:
        yield {name: record.get(name) for name in names}


pipeline = select(
    where(
        convert_types(
            parse_csv(read_lines("sales.csv")),
            {"amount": float, "quantity": int},
        ),
        lambda r: (r["amount"] or 0) > 1000,
    ),
    "region", "amount",
)

for row in pipeline:
    print(row)

Five stages, one record in flight at any moment. Each stage is a small generator that could be tested on a list of three rows.

Laziness has costs

values = (n for n in range(5))

# print(len(values))          # TypeError: no len()
# print(values[2])            # TypeError: not subscriptable
print(list(values))           # works, once
print(list(values))           # [] - exhausted

def risky():
    yield 1
    raise ValueError("failed on the second item")


gen = risky()
print("created successfully")        # no error yet
print(next(gen))                      # 1
# print(next(gen))                    # ValueError, now

A generator that will fail does not fail when created. If validation matters, validate eagerly and stream afterwards.

Making a lazy result reusable

import itertools

gen = (n * n for n in range(5))

a, b = itertools.tee(gen, 2)          # two independent iterators
print(list(a))
print(list(b))

tee buffers everything one branch has consumed but the other has not. If one branch races far ahead, the buffer grows to hold the difference - so it is not free, and on a large stream it can use as much memory as a list.

class Lazy:
    """Compute once, on first access, then remember."""

    def __init__(self, factory):
        self.factory = factory
        self._value = None
        self._done = False

    @property
    def value(self):
        if not self._done:
            print("computing...")
            self._value = self.factory()
            self._done = True
        return self._value


config = Lazy(lambda: {"loaded": True})
print("created")
print(config.value)       # computing... then the dictionary
print(config.value)       # cached

Chunking a stream

import itertools


def chunked(iterable, size):
    """Yield fixed size lists from any iterable, lazily."""
    iterator = iter(iterable)
    while chunk := list(itertools.islice(iterator, size)):
        yield chunk


for batch in chunked(range(1, 11), 4):
    print(batch)          # [1,2,3,4] [5,6,7,8] [9,10]


def process_in_batches(records, size=1000):
    """Useful when each batch is written somewhere expensive."""
    for batch in chunked(records, size):
        print(f"writing {len(batch)} records")

A worked example: counting without loading

from collections import Counter


def word_stream(paths):
    """Yield every word across several files, one at a time."""
    for path in paths:
        with open(path, encoding="utf-8") as handle:
            for line in handle:
                for word in line.lower().split():
                    cleaned = word.strip(".,!?;:\"'()")
                    if cleaned:
                        yield cleaned


def report(paths, top=10):
    counts = Counter(word_stream(paths))     # Counter consumes the stream
    total = sum(counts.values())
    print(f"{total:,} words, {len(counts):,} distinct")
    for word, count in counts.most_common(top):
        print(f"  {word:<16}{count:>6}  {count / total:.1%}")

The Counter holds one entry per distinct word, not one per word. Ten gigabytes of text with a fifty thousand word vocabulary needs a dictionary of fifty thousand entries.

When to be eager instead

# Be eager when the data is small and you need it repeatedly
records = list(read_records("small.csv"))
print(len(records))
print(records[0])
print(sorted(records, key=lambda r: r["name"])[:3])

# Be eager when the source must be closed before you finish using the data
def broken(path):
    with open(path, encoding="utf-8") as handle:
        return (line for line in handle)      # the file closes on return


# for line in broken("notes.txt"):
#     print(line)          # ValueError: I/O operation on closed file

That last trap is worth remembering. Returning a generator expression from inside a with block hands back something that reads from an already closed file. Use yield inside the block instead, so the block stays open while the generator lives.

Common mistakes

• Wrapping a lazy object in list() out of habit, discarding the benefit.
• Iterating a generator twice.
• Calling len() or indexing a lazy object.
• Returning a generator expression from inside a with block.
• Using tee on a large stream and quietly buffering all of it.
• Expecting errors inside a generator to surface when it is created.
• Sorting a lazy stream, which necessarily materialises the whole thing.

Best practices

• Stay lazy from the source until the point where a decision is made.
• Build pipelines from small single purpose generators.
• Materialise with list() only when you need length, indexing or a second pass.
• Use yield inside a with block, never return of a generator expression.
• Validate eagerly if a failure must be reported before processing starts.
• Use islice to bound anything that might be infinite.

Practice

1. Count the matching lines in a large file without loading it, and prove memory stays constant.
2. Build a four stage lazy pipeline and add a print to each stage to observe the interleaving.
3. Demonstrate the closed file trap and fix it.
4. Write a chunked function and use it to batch a stream of ten thousand records.
5. Explain why sorted(generator) cannot be lazy.

Conclusion

Lazy evaluation trades random access for constant memory and the ability to stop early. Keep data flowing through generators from the source, and materialise it only at the point where you genuinely need all of it at once.