The pipeline pattern: a practical design patterns guide
When data flows through your system, it rarely gets processed in a single step. The pipeline pattern: a practical design patterns guide treats your codebase like a factory floor, where each station has one job and passes its output to the next. By breaking complex transformations into discrete, composable steps, you gain clarity, testability, and the ability to reconfigure your processing logic without rewriting everything.
What is the pipeline pattern?
The pipeline pattern is a structural design pattern where data moves through a sequence of processing stages. Each stage—or step—receives input, applies a transformation, and produces output that becomes input for the next step. This creates a chain of responsibility where each component is oblivious to anything beyond its immediate neighbors.
In production systems, we see pipelines everywhere: image resizing services crop, then filter, then compress; ETL jobs extract data, transform it, then load it into warehouses; HTTP middleware validates requests, authenticates users, then passes control forward. The pattern works because it enforces separation of concerns and makes the flow of data explicit and traceable.
Steps with one job — Each step transforms input and passes it on
The foundation of an effective pipeline is the single-responsibility principle applied at scale. Each step should do one thing well, accept input of a known type, and emit output that the next step can consume.
Consider an e-commerce order processing pipeline. One team might write it like this:
class OrderProcessor {
async process(order: Order): Promise {
// Validate, calculate tax, check inventory, apply discounts,
// create invoice, send email, log to analytics...
// 500 lines of spaghetti
}
}
A better approach uses dedicated steps:
interface PipelineStep {
execute(input: T): Promise;
name: string;
}
class ValidateOrderStep implements PipelineStep {
name = 'ValidateOrder';
async execute(order: Order): Promise {
if (!order.customerId || order.items.length === 0) {
throw new ValidationError('Invalid order structure');
}
return order;
}
}
class CalculateTaxStep implements PipelineStep {
name = 'CalculateTax';
constructor(private taxService: TaxService) {}
async execute(order: Order): Promise {
order.tax = await this.taxService.calculate(order);
return order;
}
}
class CheckInventoryStep implements PipelineStep {
name = 'CheckInventory';
constructor(private inventory: InventoryService) {}
async execute(order: Order): Promise {
const available = await this.inventory.reserve(order.items);
if (!available) {
throw new OutOfStockError('Some items unavailable');
}
return order;
}
}
class OrderPipeline {
constructor(private steps: PipelineStep[]) {}
async execute(order: Order): Promise {
let result = order;
for (const step of this.steps) {
result = await step.execute(result);
}
return result;
}
}
Now each step is testable in isolation. You can mock the tax service for the tax step's unit test without touching inventory logic. The order flows through cleanly, transformed at each stage.
Composable — Add, remove or reorder steps easily
Real systems evolve. Today you need tax calculation; next quarter you add promotional code handling; in six months you integrate with a new fraud detection provider. A well-designed pipeline lets you plug these in without refactoring the entire processor.
Here's how composition plays out in practice:
// Standard pipeline for regular orders
const standardPipeline = new OrderPipeline([
new ValidateOrderStep(),
new CalculateTaxStep(taxService),
new CheckInventoryStep(inventory),
new ApplyDiscountStep(discountEngine),
new CreateInvoiceStep(invoiceService),
]);
// Premium customer pipeline: adds fraud check, skips some steps
const premiumPipeline = new OrderPipeline([
new ValidateOrderStep(),
new CheckFraudStep(fraudDetector), // New step
new CalculateTaxStep(taxService),
new CheckInventoryStep(inventory),
new ApplyPremiumDiscountStep(discountEngine), // Different step
new CreateInvoiceStep(invoiceService),
new SendThankYouEmailStep(emailService), // Added
]);
// Internal order pipeline: skip validation (trusted source)
const internalPipeline = new OrderPipeline([
new CalculateTaxStep(taxService),
new CheckInventoryStep(inventory),
new ApplyDiscountStep(discountEngine),
new CreateInvoiceStep(invoiceService),
]);
This flexibility is gold. You deploy different pipelines for different scenarios—A/B testing a new payment flow, handling internal orders differently, or gradually rolling out a fraud detection layer. Adding a new middleware step is a three-line change, not a refactor of your entire order processor.
Examples — Image processing, ETL, request middleware and validation
Let's walk through real-world scenarios where the pipeline pattern shines.
Image processing pipeline
In a travel platform, when a user uploads a hotel photo, you need to normalize its size, compress it, generate thumbnails, and detect if it's actually appropriate (no screenshots of charts, no logos). A pipeline handles this elegantly:
interface ImageData {
buffer: Buffer;
filename: string;
metadata?: Record;
}
class ResizeImageStep implements PipelineStep {
name = 'ResizeImage';
constructor(private sharp: SharpClient) {}
async execute(img: ImageData): Promise {
img.buffer = await this.sharp.resize(img.buffer, 1920, 1080);
return img;
}
}
class CompressImageStep implements PipelineStep {
name = 'CompressImage';
constructor(private sharp: SharpClient) {}
async execute(img: ImageData): Promise {
img.buffer = await this.sharp.jpeg(img.buffer, { quality: 80 });
return img;
}
}
class ContentModerationStep implements PipelineStep {
name = 'ContentModeration';
constructor(private moderationApi: ModerationService) {}
async execute(img: ImageData): Promise {
const result = await this.moderationApi.check(img.buffer);
if (!result.approved) {
throw new RejectedContentError(`Image flagged: ${result.reason}`);
}
img.metadata = { moderation: result };
return img;
}
}
const imagePipeline = new Pipeline([
new ResizeImageStep(sharp),
new CompressImageStep(sharp),
new ContentModerationStep(moderation),
]);
Each transformation is independent and can be tested with sample images. You can swap in a different resize algorithm or add a new step (watermarking, metadata stripping) without touching the others.
ETL pipeline
Extract, transform, load jobs benefit hugely from the pipeline pattern. Instead of writing a monolithic script that reads a database, joins tables, filters rows, and writes to a warehouse, you compose steps:
interface DataBatch {
rows: Record[];
metadata: { source: string; timestamp: number };
}
class ExtractFromDatabaseStep implements PipelineStep {
name = 'ExtractFromDatabase';
constructor(private db: Database) {}
async execute(batch: DataBatch): Promise {
batch.rows = await this.db.query('SELECT * FROM raw_events WHERE processed = false LIMIT 10000');
return batch;
}
}
class NormalizeFieldsStep implements PipelineStep {
name = 'NormalizeFields';
async execute(batch: DataBatch): Promise {
batch.rows = batch.rows.map(row => ({
...row,
email: row.email.toLowerCase(),
created_at: new Date(row.timestamp).toISOString(),
}));
return batch;
}
}
class FilterInvalidStep implements PipelineStep {
name = 'FilterInvalid';
async execute(batch: DataBatch): Promise {
batch.rows = batch.rows.filter(row =>
row.email && row.email.includes('@') && row.created_at
);
return batch;
}
}
class LoadToWarehouseStep implements PipelineStep {
name = 'LoadToWarehouse';
constructor(private warehouse: DataWarehouse) {}
async execute(batch: DataBatch): Promise {
await this.warehouse.insert('events', batch.rows);
return batch;
}
}
const etlPipeline = new Pipeline([
new ExtractFromDatabaseStep(db),
new NormalizeFieldsStep(),
new FilterInvalidStep(),
new LoadToWarehouseStep(warehouse),
]);
Each step is independently testable. You can mock the database extraction and focus on normalization logic. When a data quality issue crops up, you know exactly which step to blame and can add a debugging log or metric to that step alone.
HTTP middleware pipeline
Most web frameworks ship with middleware concepts, which are pipelines in disguise. Request and response objects flow through a chain of handlers. Even if your framework doesn't have a formal pipeline, you can model it explicitly:
interface RequestContext {
req: Request;
res: Response;
user?: User;
authenticated: boolean;
startTime: number;
}
class LoggingMiddleware implements PipelineStep {
name = 'Logging';
async execute(ctx: RequestContext): Promise {
ctx.startTime = Date.now();
console.log(`[${new Date().toISOString()}] ${ctx.req.method} ${ctx.req.path}`);
return ctx;
}
}
class AuthenticationMiddleware implements PipelineStep {
name = 'Authentication';
constructor(private tokenService: TokenService) {}
async execute(ctx: RequestContext): Promise {
const token = ctx.req.headers.authorization?.split(' ')[1];
if (!token) {
ctx.authenticated = false;
return ctx;
}
try {
ctx.user = await this.tokenService.verify(token);
ctx.authenticated = true;
} catch {
ctx.authenticated = false;
}
return ctx;
}
}
class AuthorizationMiddleware implements PipelineStep {
name = 'Authorization';
async execute(ctx: RequestContext): Promise {
if (!ctx.authenticated) {
ctx.res.status(401).json({ error: 'Unauthorized' });
throw new AuthorizationError('Not authenticated');
}
return ctx;
}
}
class RateLimitMiddleware implements PipelineStep {
name = 'RateLimit';
constructor(private limiter: RateLimiter) {}
async execute(ctx: RequestContext): Promise {
const limited = await this.limiter.check(ctx.user.id);
if (limited) {
ctx.res.status(429).json({ error: 'Too many requests' });
throw new RateLimitError('Rate limit exceeded');
}
return ctx;
}
}
This explicit pipeline makes request flow transparent and testable. You can mock the token service for auth tests or the rate limiter to verify that requests are blocked appropriately.
Observability — Log timing per step to find bottlenecks
A pipeline's superpower is visibility. When something slows down, you want to know which step is the culprit. Build timing instrumentation into the pipeline executor:
interface StepMetrics {
name: string;
durationMs: number;
status: 'success' | 'error';
error?: string;
}
class ObservablePipeline {
constructor(private steps: PipelineStep[]) {}
async execute(input: T, context?: Record): Promise<{ result: T; metrics: StepMetrics[] }> {
const metrics: StepMetrics[] = [];
let result = input;
for (const step of this.steps) {
const start = Date.now();
try {
result = await step.execute(result);
const duration = Date.now() - start;
metrics.push({ name: step.name, durationMs: duration, status: 'success' });
// Log to observability platform
console.log(`Step ${step.name} completed in ${duration}ms`);
} catch (error) {
const duration = Date.now() - start;
metrics.push({
name: step.name,
durationMs: duration,
status: 'error',
error: error.message
});
console.error(`Step ${step.name} failed after ${duration}ms: ${error.message}`);
throw error; // Stop pipeline on error
}
}
return { result, metrics };
}
}
// Usage with metrics collection
const pipeline = new ObservablePipeline([...steps]);
const { result, metrics } = await pipeline.execute(order);
// Send to DataDog or similar
for (const metric of metrics) {
datadog.gauge('pipeline.step.duration', metric.durationMs, {
tags: [`step:${metric.name}`, `status:${metric.status}`]
});
}
Now you can see at a glance: is the tax service slow? Is inventory check timing out? This transparency is invaluable in production debugging. You can set alerts on slow steps or collect percentiles (p95, p99) to spot degradation over time.
Common mistakes
Mixing concerns within steps. A step that validates, transforms, and logs is doing three jobs. Keep steps pure: one transformation, minimal side effects. Logging belongs in the pipeline executor, not in steps.
Tight coupling between steps. If step B assumes step A ran before it and set a property, you've created hidden dependencies. Each step should be independently useful. Document input and output contracts clearly.
No error handling strategy. What happens when a step fails? Does the entire pipeline halt (fast-fail)? Do you skip that step and continue? Do you collect errors and log them? Decide upfront and implement it consistently.
Ignoring performance in observability. Adding metrics to pipelines is cheap and pays for itself in production. If you skip this, you'll debug blind when things slow down.
Over-engineering for "flexibility." If you genuinely have one pipeline that never changes, don't build an elaborate, generic composition system. Start simple and refactor toward the pattern when you need multiple variants.
Key takeaways
- The pipeline pattern breaks complex data transformations into discrete, single-purpose steps that each receive input and produce output for the next step.
- Each step should have one job, be testable in isolation, and maintain no hidden dependencies on other steps.
- Pipelines become composable: you can add, remove, or reorder steps to support different workflows without rewriting the core logic.
- Real-world pipelines—image processing, ETL, HTTP middleware, validation chains—all benefit from this structural clarity and testability.
- Instrument pipelines with per-step timing metrics to identify bottlenecks and catch performance regressions early.
- Keep steps pure and side-effect-free; move logging and cross-cutting concerns into the pipeline executor.
Comments
Loading comments…