TopolearnLog analysis, ETL and data cleaning over huge inputs stay fast and memory-light when each stage is lazy.
A pipeline is a series of lazy stages, each taking an iterator and returning another. Data flows through one item at a time, so memory stays constant no matter how large the input:
lines = open(path)
records = (parse(l) for l in lines if l.strip())
errors = (r for r in records if r.level == "ERROR")
first20 = islice(errors, 20)Nothing runs until the last stage is consumed. If you stop early (here, after 20 errors), the earlier stages stop too, so most of the file is never read.
Design tips:
- Make each stage a small generator function or generator expression with one job: read, parse, filter, transform, batch.
- Use itertools for structural steps:
chain.from_iterableto merge sources,isliceto limit,groupbyfor sorted runs,batchedto write in chunks. - Keep side effects (writing output, network calls) at the end of the pipeline.
- Close resources: wrap file opening in a generator using
with, so the file closes when the generator finishes or is garbage-collected, or consume within awithblock.
Laziness makes errors appear late: a parse error surfaces when the record is reached, not when the pipeline is built. Include line numbers (via enumerate) in records to make errors traceable.
Example
from itertools import batched, chain, islice
logs = {"a.log": "INFO start\nERROR disk\n\nINFO ok\n",
"b.log": "ERROR net\nERROR disk\nINFO bye\n"}
for name, text in logs.items():
open(name, "w").write(text)
def read_lines(path):
with open(path) as f: # closed when the generator finishes
for n, line in enumerate(f, 1):
yield path, n, line.rstrip("\n")
lines = chain.from_iterable(read_lines(p) for p in logs)
records = ((p, n, *l.split(" ", 1)) for p, n, l in lines if l)
errors = (r for r in records if r[2] == "ERROR")
for batch in batched(islice(errors, 3), 2):
print(batch)
# (('a.log', 2, 'ERROR', 'disk'), ('b.log', 1, 'ERROR', 'net'))
# (('b.log', 2, 'ERROR', 'disk'),)Remember
- Each pipeline stage is a lazy iterator; memory stays constant
- Stopping the last stage early stops all earlier stages
- Keep stages small and carry positions for traceable errors
Watch out for
- Inserting a
list()in the middle of a pipeline, which loads everything into memory and defeats laziness.