When processing large datasets, it's common to chain together multiple operations such as reading data, filtering records, transforming values, and writing the results somewhere else.
A straightforward implementation often creates intermediate lists after each step:
data = read_data()
filtered = filter_data(data)
transformed = transform_data(filtered)
save(transformed)
While this works well for small datasets, it becomes inefficient when processing millions of records because every intermediate result is stored in memory.
A better approach is to use generator-based pipelines.
Generators allow data to flow through each processing stage one item at a time, making your code more memory-efficient, composable, and easier to maintain.
In this article, we'll build several generator pipelines ranging from simple text processing to ETL and computer vision workloads.
What Is a Generator Pipeline?
A generator pipeline consists of multiple independent stages.
Each stage:
- Receives an iterable (usually another generator).
- Processes each item.
- Produces items using
yield.
The output of one stage becomes the input of the next.
One of the biggest advantages is that only one item typically moves through the pipeline at a time.
Example 1: Processing a Large Text File
Suppose you have a multi-gigabyte log file and want to:
- Ignore comments
- Remove whitespace
- Convert everything to uppercase
Each step becomes its own generator.
def read_lines(filename):
with open(filename) as f:
for line in f:
yield line
def remove_comments(lines):
for line in lines:
if not line.startswith("#"):
yield line
def strip_lines(lines):
for line in lines:
yield line.strip()
def uppercase(lines):
for line in lines:
yield line.upper()
Connecting them is straightforward:
pipeline = uppercase(
strip_lines(
remove_comments(
read_lines("log.txt")
)
)
)
for line in pipeline:
print(line)
Notice that nothing happens until the final loop starts consuming the generator.
At any point, only a single line is being processed.
Example 2: Building a Simple ETL Pipeline
Generator pipelines are a natural fit for ETL (Extract, Transform, Load).
Let's define a simple model.
from dataclasses import dataclass
@dataclass
class User:
id: int
name: str
age: int
Extract
def extract():
users = [
{"id": 1, "name": "Alice", "age": 22},
{"id": 2, "name": "Bob", "age": 17},
{"id": 3, "name": "Charlie", "age": 35},
]
for user in users:
yield user
Filter
def filter_adults(users):
for user in users:
if user["age"] >= 18:
yield user
Transform
def transform(users):
for user in users:
yield User(**user)
Load
def load(users):
for user in users:
print(user)
Putting Everything Together
load(
transform(
filter_adults(
extract()
)
)
)
Output:
User(id=1, name='Alice', age=22)
User(id=3, name='Charlie', age=35)
Each stage has a single responsibility, making the pipeline easy to extend and test.
Example 3: Computer Vision Pipeline
Generator pipelines are especially useful in machine learning and computer vision, where images can consume significant memory.
def load_images(paths):
for path in paths:
image = cv2.imread(path)
yield image
def resize(images):
for image in images:
yield cv2.resize(image, (224, 224))
def normalize(images):
for image in images:
yield image / 255.0
def predict(images, model):
for image in images:
yield model.predict(image[None])
Usage:
predictions = predict(
normalize(
resize(
load_images(image_paths)
)
),
model,
)
for prediction in predictions:
...
Instead of loading every image into memory first, images are loaded, processed, and predicted one at a time.
Example 4: Processing Database Records
The same approach works for database queries.
def fetch_rows(cursor):
while True:
rows = cursor.fetchmany(1000)
if not rows:
break
for row in rows:
yield row
Cleaning the records:
def clean(rows):
for row in rows:
row["name"] = row["name"].strip()
yield row
Saving them:
def save(rows):
for row in rows:
insert_into_new_db(row)
Only small batches are fetched from the database, while individual rows continue flowing through the pipeline.
Using yield from
Sometimes a generator simply forwards another iterable.
Instead of manually iterating:
def flatten(data):
for lst in data:
for item in lst:
yield item
You can delegate using yield from.
def flatten(data):
for lst in data:
yield from lst
Example:
data = [[1, 2], [3, 4], [5]]
print(list(flatten(data)))
Output:
[1, 2, 3, 4, 5]
yield from makes generator composition cleaner and more readable.
Generator Expressions
For simple transformations, generator expressions are concise and expressive.
numbers = range(10)
pipeline = (
x * x
for x in numbers
if x % 2 == 0
)
print(list(pipeline))
Output:
[0, 4, 16, 36, 64]
They work particularly well for lightweight pipelines.
Building Configurable Pipelines
As applications grow, it's useful to compose stages dynamically.
from collections.abc import Iterable, Callable
def run_pipeline(data: Iterable, *stages: Callable):
for stage in stages:
data = stage(data)
return data
Now define a pipeline at runtime.
pipeline = run_pipeline(
range(20),
lambda xs: (x for x in xs if x % 2 == 0),
lambda xs: (x * x for x in xs),
lambda xs: (x + 1 for x in xs),
)
print(list(pipeline))
Output:
[1, 5, 17, 37, 65, 101, 145, 197, 257, 325]
This makes it easy to swap, reorder, or add processing stages without modifying existing code.
Why Use Generator Pipelines?
Generator pipelines provide several advantages over list-based processing.
- Memory efficient — process one item at a time instead of storing intermediate collections.
- Lazy evaluation — work happens only when data is requested.
- Composable — each stage performs a single responsibility.
- Reusable — stages can be mixed and matched across pipelines.
- Scalable — ideal for files, database cursors, streams, APIs, and machine learning datasets.
- Testable — each generator can be unit tested independently.
Where You'll See This Pattern
Generator pipelines appear in many real-world systems, including:
- ETL and data engineering workflows
- Log processing
- Web scraping
- CSV and JSON processing
- Computer vision preprocessing
- Machine learning data loaders
- Streaming applications
- API response processing
Once you start recognizing the pattern, you'll find opportunities to use it in many Python projects.
Conclusion
Python generators are more than a way to iterate over data—they're a powerful tool for building clean, modular, and memory-efficient processing pipelines.
By splitting your workflow into small generator stages, you can process datasets of virtually any size while keeping your code easy to understand, test, and extend.
If you're working with large files, databases, APIs, or AI workloads, generator pipelines are a technique worth adding to your Python toolbox.
Happy coding! 🚀
Top comments (0)