Lazy Evaluation Transform Lineage Mock in Python
Build a mock lineage tracker for data transforms using lazy evaluation and function wrappers in Python.
Python code
48 linesimport 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
[{'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
- Use a class with `__call__` to track lineage more explicitly
- 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
More from Big data & Spark
Keep learning
Related tutorials and quizzes for this topic.