Most data engineers meet design patterns in a CS course and never connect them to the actual job. But the messes that slow a data team down - the same connection code written three times, an 800-line transform full of nested conditionals, a new source that means copy-pasting from five files - are pattern problems, not skill problems. Here are five patterns that map directly onto data work, each with the problem it solves, a tight example, and an honest note on when not to reach for it.

1. Factory Method - flexible source connectors

Your pipeline reads from Postgres today, MongoDB tomorrow, S3 next week, and the orchestration code fills up with if source_type == ... branches. A factory moves that decision to one place behind a common interface, so the rest of the code just asks for a connector and uses it.

from abc import ABC, abstractmethod

class DataConnector(ABC):
    @abstractmethod
    def read(self, query: str): ...

class PostgresConnector(DataConnector):
    def read(self, query): ...   # returns a DataFrame

class S3Connector(DataConnector):
    def read(self, key): ...

def make_connector(cfg: dict) -> DataConnector:
    return {"postgres": PostgresConnector,
            "s3": S3Connector}[cfg["type"]](**cfg["params"])

Use it when you support several interchangeable sources. Skip it when there's one source and likely always will be - the indirection is just overhead.

2. Builder - constructing complex transformations

A transformation that takes ten optional parameters becomes an unreadable constructor call, and order starts to matter in ways nobody documents. A builder assembles the object step by step with named, chainable calls.

class PipelineBuilder:
    def __init__(self): self._steps = []
    def filter(self, cond): self._steps.append(("filter", cond)); return self
    def dedupe(self, keys): self._steps.append(("dedupe", keys)); return self
    def build(self): return Pipeline(self._steps)

pipeline = (PipelineBuilder()
            .filter("amount > 0")
            .dedupe(["id"])
            .build())

Use it for objects with many optional parts. Skip it for two or three arguments - a plain function is clearer.

3. Decorator - observability without touching core logic

You want timing, logging, and retries around your transform functions without burying the business logic in instrumentation. A decorator wraps behavior around a function, leaving its body untouched.

import time, functools, logging

def timed(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        logging.info("%s took %.2fs", fn.__name__, time.perf_counter() - start)
        return result
    return wrapper

@timed
def transform(df): ...

Use it for cross-cutting concerns - timing, logging, retries, caching. Skip it when the wrapper hides control flow a reader needs to see.

4. Adapter - wrapping legacy systems and external APIs

A third-party API or a legacy system speaks a shape your pipeline doesn't. Rather than litter your code with its quirks, an adapter translates it to the interface the rest of your code already expects.

class LegacyExportClient:
    def fetch_xml(self): ...   # ugly, not yours

class LegacySourceAdapter(DataConnector):
    def __init__(self, client): self._client = client
    def read(self, _query=None):
        return xml_to_dataframe(self._client.fetch_xml())

Use it to isolate code you don't control behind a clean boundary. Skip it when you can call the API directly without the impedance mismatch.

5. Abstract Factory - families of related objects

When a whole set of objects must vary together - say a connector, a writer, and a schema validator that all have to match "Snowflake" or all match "BigQuery" - an abstract factory produces a consistent family so you can't accidentally mix a Snowflake reader with a BigQuery writer.

class WarehouseFactory(ABC):
    @abstractmethod
    def reader(self) -> DataConnector: ...
    @abstractmethod
    def writer(self): ...

class SnowflakeFactory(WarehouseFactory):
    def reader(self): return SnowflakeConnector()
    def writer(self): return SnowflakeWriter()

Use it when objects come in matched sets that must stay consistent. Skip it if your objects vary independently - it adds structure you won't use.

The point isn't the patterns

Patterns are a vocabulary, not a goal. The reason to know them is that the next time you feel a pipeline getting brittle, you recognize the shape - "this is a factory problem" - and reach for a known fix instead of another nested conditional. Use them where they remove duplication and clarify intent; resist them where they add ceremony to code that was already simple. The best pattern is often the one you didn't need.