A data pipeline splits a complex transformation into small stages and passes each stage’s output to the next one.
What is a Data Pipeline?
A Data Pipeline is a sequential chain of functions where the output of one becomes the input of the next.
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Without pipeline: intermediate variables
even = [x for x in numbers if x % 2 == 0]
doubled = [x * 2 for x in even]
total = sum(doubled)
print(total) # 60
# With pipeline: one chain
from functools import reduce
total = reduce(
lambda acc, x: acc + x,
map(lambda x: x * 2, filter(lambda x: x % 2 == 0, numbers)),
0
)
print(total) # 60
Visualization
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
↓ filter(even)
[2, 4, 6, 8, 10]
↓ map(double)
[4, 8, 12, 16, 20]
↓ reduce(sum)
60
Basic Pipeline: filter → map → reduce
from functools import reduce
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Sum of squares of even numbers
result = reduce(
lambda acc, x: acc + x, # Step 3: sum
map(
lambda x: x ** 2, # Step 2: square
filter(lambda x: x % 2 == 0, numbers) # Step 1: even
),
0
)
print(result) # 220
# Even: [2, 4, 6, 8, 10]
# Squares: [4, 16, 36, 64, 100]
# Sum: 220
The same pipeline is more readable with a list comprehension:
result = sum([x ** 2 for x in numbers if x % 2 == 0])
print(result) # 220
The pipe() Function
A general-purpose function for building pipelines:
def pipe(data, *functions):
"""Apply functions sequentially."""
result = data
for func in functions:
result = func(result)
return result
# Define steps
def filter_even(numbers):
return [x for x in numbers if x % 2 == 0]
def double_all(numbers):
return [x * 2 for x in numbers]
def sum_all(numbers):
return sum(numbers)
# Run the pipeline
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
result = pipe(numbers, filter_even, double_all, sum_all)
print(result) # 60
Each step is a separate function — easy to add, remove, or reorder.
Practical Example: Order Processing
from functools import reduce
data = [
{"product": "Phone", "quantity": 2, "price": 500},
{"product": "Laptop", "quantity": 1, "price": 1200},
{"product": "Mouse", "quantity": 5, "price": 25},
{"product": "Monitor", "quantity": 2, "price": 300}
]
# Pipeline: quantity > 1 → compute total → sum
filtered = filter(lambda item: item["quantity"] > 1, data)
totals = map(lambda item: item["quantity"] * item["price"], filtered)
grand_total = reduce(lambda acc, x: acc + x, totals, 0)
print(grand_total) # 1725
# Phone: 2 × 500 = 1000
# Mouse: 5 × 25 = 125
# Monitor: 2 × 300 = 600
# Total: 1725
Pipeline Class
A fluent interface for convenient chaining:
class Pipeline:
def __init__(self, data):
self.data = data
def filter(self, predicate):
self.data = [x for x in self.data if predicate(x)]
return self
def map(self, transform):
self.data = [transform(x) for x in self.data]
return self
def reduce(self, reducer, initial=None):
from functools import reduce
if initial is None:
return reduce(reducer, self.data)
return reduce(reducer, self.data, initial)
def collect(self):
return self.data
# Usage
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
result = (Pipeline(numbers)
.filter(lambda x: x % 2 == 0)
.map(lambda x: x * 2)
.reduce(lambda acc, x: acc + x, 0))
print(result) # 60
Common Mistakes
Mistake 1: Mutating the original data
# ❌ Bad: sort() modifies the original list
def bad_pipeline(data):
data.sort()
return [x * 2 for x in data]
numbers = [3, 1, 2]
result = bad_pipeline(numbers)
print(numbers) # [1, 2, 3] ← changed!
# ✅ Good: sorted() creates a new list
def good_pipeline(data):
sorted_data = sorted(data)
return [x * 2 for x in sorted_data]
numbers = [3, 1, 2]
result = good_pipeline(numbers)
print(numbers) # [3, 1, 2] ← unchanged
Mistake 2: Deeply nested pipelines
# ❌ Unreadable
result = reduce(
lambda a, x: a + x,
map(lambda x: x ** 2,
filter(lambda x: x > 0,
map(lambda x: x - 10,
filter(lambda x: x % 2 == 0, data)))),
0
)
# ✅ Break into named steps
step1 = [x for x in data if x % 2 == 0]
step2 = [x - 10 for x in step1]
step3 = [x for x in step2 if x > 0]
result = sum(x ** 2 for x in step3)
Good Pipeline Rules
- Each step is a pure function (does not modify input data)
- Break complex pipelines into named steps
- Use list comprehensions where they are more readable
- Use generators instead of lists for large datasets
💬 Comments (0)
No comments yet
Be the first to share your opinion about this article!