Build a Generator Pipeline
A generator pipeline chains small `yield`-based stages so data flows through them one item at a time. The result is constant-memory streaming over inputs that would not fit in RAM, with each stage doing one job (read, parse, filter, transform, sink). This entry shows a three-stage pipeline, how to compose stages dynamically, and why generator pipelines beat list-of-lists processing for log-style data.
209 views
5
def lines(source):
"""Stage 1: yield each raw line from a list of strings."""
for line in source:
yield line
def parse_numbers(stream):
"""Stage 2: turn each line into an int (skip non-numeric lines)."""
for line in stream:
line = line.strip()
if line.lstrip('-').isdigit():
yield int(line)
def positive_only(stream):
"""Stage 3: keep only the positive numbers."""
for value in stream:
if value > 0:
yield value
raw = ['12', ' -3', 'oops', '0', ' 77 ', '', '5']
pipeline = positive_only(parse_numbers(lines(raw)))
# Nothing has run yet. The first next() drives the whole pipeline by one item.
print(list(pipeline)) # [12, 77, 5]Each stage is a generator: a function with yield that produces one item per call. Wrapping lines inside parse_numbers inside positive_only does not run anything; it just builds a chain of iterators. The first next() (or list(...)) walks the chain backwards: the sink asks for one item, the filter asks the parser, the parser asks the source. Because only one item is in flight at a time, the pipeline streams arbitrarily large inputs in O(1) memory. The same shape scales to ETL jobs that read 10 GB of CSV without ever holding more than a row.
from functools import reduce
def map_stage(fn):
def stage(stream):
for item in stream:
yield fn(item)
return stage
def filter_stage(pred):
def stage(stream):
for item in stream:
if pred(item):
yield item
return stage
stages = [
map_stage(str.strip),
filter_stage(lambda s: s.lstrip('-').isdigit()),
map_stage(int),
filter_stage(lambda n: n > 0),
map_stage(lambda n: n * 2),
]
raw = ['12', ' -3', 'oops', '0', ' 77 ', '', '5']
result = reduce(lambda stream, stage: stage(stream), stages, iter(raw))
print(list(result)) # [24, 154, 10]Generator stages compose by function application: each stage wraps the previous one. reduce(lambda s, stage: stage(s), stages, source) threads the iterator through every stage left-to-right, which is exactly what a Unix pipe does. map_stage and filter_stage are tiny factories that turn any predicate or transform into a reusable pipeline node. This style is the right tool when stage order is configuration-driven (CLI flags, JSON pipeline definitions) rather than hard-coded function nesting.
# Eager: every stage materializes a full intermediate list.
def eager_double_positive_evens(values):
step1 = [v for v in values if v > 0]
step2 = [v for v in step1 if v % 2 == 0]
step3 = [v * 2 for v in step2]
return step3
# Lazy: each stage is a generator. No intermediate lists at all.
def lazy_double_positive_evens(values):
positive = (v for v in values if v > 0)
even = (v for v in positive if v % 2 == 0)
doubled = (v * 2 for v in even)
yield from doubled
import itertools
big_source = range(-1000, 1000)
# Both produce the same answer, but the lazy version never builds intermediates.
print(eager_double_positive_evens(big_source)[:5])
print(list(itertools.islice(lazy_double_positive_evens(big_source), 5)))
# [4, 8, 12, 16, 20]
# [4, 8, 12, 16, 20]
# Short-circuit: stop after the first 5. Eager paid for all 1000+ items first.
# Lazy stopped after producing exactly 5.The eager version allocates three intermediate lists; for million-row inputs that is three full copies in memory at once. The lazy version replaces every list comprehension with a generator expression (... for ...), so the pipeline is end-to-end streaming. Pair it with itertools.islice (or any consumer that breaks early) and you only pay for the items you actually consume. The mental model is 'pipes, not arrays': data flows, it does not pile up.