Build a Data Processing Pipeline in Python

Every concept in this section on iterators and generators has been building toward a practical skill: the ability to design and build a data processing pipeline that reads input, transforms it through multiple stages, and writes output, all without storing more than a single item in memory at any time. A well-designed generator pipeline replaces hundreds of lines of nested loops and intermediate lists with a chain of small, focused functions, each doing one job and yielding to the next. The result is code that is easier to write, easier to test, easier to debug, and capable of processing datasets of any size.

If you have worked through the articles on creating generator functions with yield and lazy evaluation in Python, you have all the building blocks. This article walks through building a complete pipeline from scratch, starting with raw input and ending with formatted output, using only generators and standard library tools. The example pipeline processes a log file, parses structured data from each line, filters out irrelevant entries, transforms the surviving data, computes aggregations, and writes a summary report, all in a single pass with no intermediate lists.

Pipeline architecture

A generator pipeline is a chain of functions where each function is a generator that takes an iterable as input and yields transformed items as output. The first function in the chain reads from a data source. The last function writes to a destination or returns a final result. Every function in between transforms the data in one specific way. The architecture follows the Unix pipes philosophy: each stage does one thing, and stages are connected by passing the output of one as the input to the next.

The key principle is that each stage should be independently testable. You should be able to call any stage with a small hand-crafted input and verify that it produces the expected output, without running the entire pipeline. This means each generator function should not know about the stages before or after it. It receives an iterable, does its job, and yields results. Where the iterable comes from and where the results go are not its concern.

A typical pipeline has these stages, though not every pipeline needs all of them. The source stage reads raw data from a file, network, or database. The parse stage converts raw text or bytes into structured records. The validate stage checks that records are well-formed and discards or flags invalid ones. The filter stage removes records that do not match business criteria. The transform stage computes derived values or normalizes formats. The aggregate stage computes summaries like counts, averages, or groupings. The sink stage writes results to a file, database, or API. Each of these stages can be a generator function that yields one record at a time.

Building the pipeline step by step

The source stage reads raw lines from a file and yields them one at a time. The parse stage receives those lines and converts them into dictionaries with named fields, skipping malformed lines:

python

def read_logs(path):
    with open(path) as f:
        for line in f:
            yield line.strip()
def parse_lines(lines):
    for line in lines:
        parts = line.split(",", 3)
        if len(parts) == 4:
            yield {
                "timestamp": parts[0],
                "level": parts[1],
                "source": parts[2],
                "message": parts[3],
            }

The filter stage keeps only records that match a condition, and the transform stage adds computed fields like extracting the hour from a timestamp:

python

def filter_errors(records):
    for rec in records:
        if rec["level"] in ("ERROR", "CRITICAL"):
            yield rec
 
def add_hour(records):
    for rec in records:
        hour = rec["timestamp"].split(":")[0][-2:]
        rec["hour"] = int(hour)
        yield rec

The aggregate stage collects statistics across all records. Unlike the previous stages, an aggregator must consume the entire input before it can produce output. It is still a generator, but it yields only after processing all records:

python

def count_by_hour(records):
    counts = {}
    for rec in records:
        hour = rec["hour"]
        counts[hour] = counts.get(hour, 0) + 1
    for hour, count in sorted(counts.items()):
        yield f"Hour {hour:02d}: {count} errors"

The pipeline is composed by passing the output of each function as input to the next. A small helper function chains generators together so the composition reads left to right in the order data flows:

python

def pipeline(source, *stages):
    result = source
    for stage in stages:
        result = stage(result)
    return result
 
report = pipeline(read_logs("server.log"), parse_lines,
                  filter_errors, add_hour, count_by_hour)
for line in report:
    print(line)

Testing pipeline stages

Each stage is a generator function that takes an iterable and returns a generator. This means you can test any stage in isolation by passing it a list of hand-crafted input records and asserting on the output. You do not need to set up a log file, a database, or a network connection to test the parsing or filtering logic:

python

def test_parse_lines():
    lines = [
        "2026-07-11 10:30,ERROR,auth,invalid password",
        "2026-07-11 10:31,INFO,server,connection accepted",
        "bad-line",
    ]
    result = list(parse_lines(lines))
    assert len(result) == 2
    assert result[0]["level"] == "ERROR"

The test creates a list of input strings, including an intentionally malformed line, and verifies that parse_lines produces the correct number of records with the correct fields. No file I/O is involved. The same pattern works for every stage in the pipeline. Filter stages are tested by checking that only matching records survive. Transform stages are tested by verifying the computed fields. Aggregate stages are tested by checking the summary output against known input.

This testability is one of the strongest arguments for the pipeline architecture. A monolithic function that reads a file, parses, filters, transforms, and writes output in one block is difficult to test because each part depends on the parts before it. A pipeline of independent generators can be tested piece by piece, and a failure in one test points directly to the stage that broke.

Error handling in pipelines

Real-world data is messy. Files have malformed lines, network connections drop, and unexpected values appear in fields that should always be numeric. A robust pipeline handles these errors without crashing and without losing the valid data that surrounds the problematic item.

The simplest error-handling strategy is to catch exceptions inside each generator stage and skip the problematic item. The parse stage can catch ValueError when converting strings to numbers and yield nothing for that item, effectively dropping it from the pipeline. The tradeoff is that silently dropping data can hide systemic problems. A better strategy logs the error and continues:

python

import logging
 
def safe_parse(lines):
    for line in lines:
        try:
            parts = line.split(",", 3)
            yield {"timestamp": parts[0], "level": parts[1]}
        except (IndexError, ValueError) as e:
            logging.warning("Skipping malformed line: %s", e)

The error is logged with enough context to investigate later, and the pipeline continues processing the next line. The valid data flows through unimpeded. This pattern of catching, logging, and continuing is the standard approach for production data pipelines that must process large volumes of real-world data.

When to use a pipeline versus a single function

Not every data processing task needs a pipeline of generators. If the task is reading a small file, applying one transformation, and writing the result, a single function with a for loop is simpler and clearer than a pipeline of four generators. The pipeline architecture pays off when the processing involves multiple distinct steps, when the data is too large for lists, or when you need to reuse individual stages in different combinations.

A sign that you should break a function into pipeline stages is when you find yourself writing comments that label sections of the function, like "parse the line", "filter out test data", or "compute the average". Each of those labelled sections is a candidate for its own generator function. Extracting them makes the code more testable and makes the data flow explicit in the function names rather than implicit in the control flow.

Another sign is when you need to reuse a transformation in multiple places. If two different scripts both need to parse log lines into structured records, the parse stage should be a standalone generator function that both scripts import. The pipeline architecture naturally encourages this kind of reuse because each stage is a self-contained function with a single responsibility.