Lazy Evaluation Transform Lineage Mock in Python

Build a mock lineage tracker for data transforms using lazy evaluation and function wrappers in Python.

Medium Python 3.9+ Aug 9, 2026 Big data & Spark 16 views 0 copies

Python code

48 lines
Python 3.9+
import functools


def lazy_transform(pipeline):
    """Build a mock lineage tracker using lazy evaluation."""
    lineage = []

    def wrap(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            lineage.append({"transform": func.__name__, "args": args, "output": result})
            return result
        return wrapper

    for name, func in pipeline:
        pipeline[name] = wrap(func)

    def execute():
        lineage.clear()
        for name, func in pipeline.items():
            func()
        return list(lineage)

    return execute


def load_data():
    return [1, 2, 3]


def filter_positive(data):
    return [x for x in data if x > 0]


def double(data):
    return [x * 2 for x in data]


if __name__ == "__main__":
    pipeline = {
        "load": load_data,
        "filter": filter_positive,
        "double": double,
    }
    run = lazy_transform(pipeline)
    result = run()
    print(result)

Output

stdout
[{'transform': 'load_data', 'args': (), 'output': [1, 2, 3]}, {'transform': 'filter_positive', 'args': (), 'output': [1, 2, 3]}, {'transform': 'double', 'args': (), 'output': [2, 4, 6]}]

How it works

The lazy_transform function wraps each pipeline function so that calling it records the function name, arguments, and output into a shared list. The wrapper uses functools.wraps to preserve metadata and executes the original function before logging. The execute closure clears the lineage and runs each transform in order, returning the accumulated log. This demonstrates lazy evaluation because the wrapping happens at definition time, but actual execution and logging occur only when execute is called.

Common mistakes

  • Forgetting to clear the lineage list before each run, causing stale logs
  • Not using `functools.wraps`, losing function metadata like `__name__`

Variations

  1. Use a class with `__call__` to track lineage more explicitly
  2. Return a list of dictionaries with timestamps for richer audit trails

Real-world use cases

  • Auditing data pipeline steps in a Spark job to track which transforms produced a given output.
  • Building a test mock that records calls to data processing functions for verification.
  • Debugging complex ETL flows by logging each transform's input and output sizes.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Big data & Spark

Related tutorials and quizzes for this topic.