Why Data Pipelines Fail
We recently reviewed a data pipeline that processed insurance claims in real-time. It ran for 47 days without issues, then suddenly stopped producing output. Nobody noticed for 6 hours. By the time anyone realized something was wrong, 15,000 claims were stuck in limbo.
The investigation revealed: a database connection pool exhausted itself. The pipeline kept opening connections and never closing them. After 47 days (which is how long the pool's lifecycle settings allowed), it ran out of connections and hung silently.
This is the norm, not the exception. Data pipelines are fragile because they sit at the boundary between your application code (which you test) and external systems (databases, APIs, message queues) which can fail in ways you never anticipated.
The good news: there are patterns that work.
Pattern 1: Idempotency First
The Problem: Your pipeline crashes halfway through processing a batch. When you restart it, do you reprocess the batch (and risk double-processing records)? Skip it (and lose data)? Nobody knows.
The Solution: Make every operation idempotent. Processing the same record twice should produce the same result as processing it once.
class IdempotentPipeline:
def __init__(self, db):
self.db = db
def process_record(self, source_record):
"""Process a record idempotently."""
# Create a stable, deterministic key
idempotency_key = hashlib.sha256(
source_record.encode()
).hexdigest()
# Check if already processed
existing = self.db.query("""
SELECT output_record FROM processed_records
WHERE idempotency_key = %s
""", [idempotency_key])
if existing:
# Already processed—return cached result
return existing[0]['output_record']
# Process the record
output = expensive_transformation(source_record)
# Store both the key and output
self.db.execute("""
INSERT INTO processed_records
(idempotency_key, source_record, output_record, timestamp)
VALUES (%s, %s, %s, %s)
""", [idempotency_key, source_record, output, now()])
return output
# Usage
pipeline = IdempotentPipeline(db)
# This is safe to call multiple times
result = pipeline.process_record("{ ... }")
result = pipeline.process_record("{ ... }") # Returns cached result
Now your pipeline is bulletproof: if it crashes and restarts, it just skips the records it already processed.
Pattern 2: Checkpoints and Resumption
The Problem: Your pipeline processes 1 million records in a batch. It fails on record 847,392. When you restart, does it start from the beginning or from record 847,393? How do you even know which record it failed on?
The Solution: Checkpoint your progress:
class CheckpointedPipeline:
def __init__(self, db, pipeline_name):
self.db = db
self.pipeline_name = pipeline_name
def get_checkpoint(self):
"""Get the last successfully processed record."""
result = self.db.query("""
SELECT checkpoint_value FROM pipeline_checkpoints
WHERE pipeline_name = %s
ORDER BY timestamp DESC
LIMIT 1
""", [self.pipeline_name])
return result[0]['checkpoint_value'] if result else 0
def save_checkpoint(self, checkpoint_value):
"""Save progress."""
self.db.execute("""
INSERT INTO pipeline_checkpoints
(pipeline_name, checkpoint_value, timestamp)
VALUES (%s, %s, %s)
""", [self.pipeline_name, checkpoint_value, now()])
def process_batch(self, records):
"""Process from last checkpoint."""
start_idx = self.get_checkpoint()
for idx, record in enumerate(records[start_idx:], start=start_idx):
try:
self.process_record(record)
self.save_checkpoint(idx)
except Exception as e:
logger.error(f"Failed at record {idx}: {e}")
raise # Stop here; restart will resume from checkpoint
# Usage
pipeline = CheckpointedPipeline(db, "claims_processing")
pipeline.process_batch(all_claims)
# If it crashes at record 847,392, restarting picks up from 847,393
Now when your pipeline crashes, restarting it automatically resumes from where it left off. No manual intervention needed.
Pattern 3: Observability at Every Step
The Problem: Your pipeline runs silently. You don't know how many records it processed, how many it skipped, how many failed. You just know it "ran" because the process exited with code 0.
The Solution: Emit metrics at every step:
from prometheus_client import Counter, Histogram, Gauge
import time
class ObservablePipeline:
def __init__(self, db, pipeline_name):
self.db = db
self.pipeline_name = pipeline_name
# Metrics
self.records_processed = Counter(
'pipeline_records_processed_total',
'Total records processed',
['pipeline_name']
)
self.records_failed = Counter(
'pipeline_records_failed_total',
'Total records that failed',
['pipeline_name']
)
self.processing_time = Histogram(
'pipeline_processing_time_seconds',
'Time to process a record',
['pipeline_name']
)
self.pipeline_lag = Gauge(
'pipeline_lag_seconds',
'How far behind we are',
['pipeline_name']
)
def process_record(self, record):
"""Process with observability."""
start = time.time()
try:
result = self.do_processing(record)
self.records_processed.labels(self.pipeline_name).inc()
return result
except Exception as e:
self.records_failed.labels(self.pipeline_name).inc()
logger.error(f"Record failed: {e}")
raise
finally:
duration = time.time() - start
self.processing_time.labels(self.pipeline_name).observe(duration)
def get_pipeline_lag(self):
"""How far behind are we?"""
latest_source = self.db.query("""
SELECT MAX(timestamp) as latest FROM source_data
""")[0]['latest']
latest_processed = self.db.query(f"""
SELECT MAX(timestamp) as latest FROM processed_data
""")[0]['latest']
lag = (latest_source - latest_processed).total_seconds()
self.pipeline_lag.labels(self.pipeline_name).set(lag)
return lag
# In your alerting:
# Alert if records_failed_total increased
# Alert if pipeline_lag > 3600 (more than 1 hour behind)
# Alert if processing_time_seconds > 5 (single records taking too long)
Now your monitoring dashboard shows:
- How many records have been processed
- How many have failed
- How far behind you are
- How long each record takes to process
When something breaks, you see it immediately.
Pattern 4: Isolated Failure Domains
The Problem: Your pipeline calls an external API. The API goes down. Your pipeline crashes and stops processing all records, even ones that don't depend on that API.
The Solution: Isolate failures:
class ResilientPipeline:
def __init__(self, db):
self.db = db
self.api_client = APIClient()
def enrich_record(self, record):
"""Enrich a record with external data."""
# This can fail; that's OK
try:
enrichment = self.api_client.lookup(record['id'])
except APIError as e:
logger.warning(f"API lookup failed for {record['id']}: {e}")
enrichment = None # Continue without enrichment
# Process the record regardless
processed = {
**record,
'enrichment': enrichment,
'enrichment_status': 'success' if enrichment else 'skipped'
}
return processed
# Result: API goes down? Fine, records still get processed.
# They're just missing enrichment. You can retry enrichment later.
The key insight: your pipeline should have a core path (record → database) and optional paths (record → API enrichment). If an optional path fails, the core path continues.
Pattern 5: Dead Letter Queues
The Problem: A record fails processing. It's not your fault—the record itself is malformed. You can't skip it (audit trail) but it's blocking the pipeline.
The Solution: Dead letter queues:
class PipelineWithDLQ:
def __init__(self, db):
self.db = db
self.dlq = "failed_records_queue"
def process_with_dlq(self, record):
"""Process, with fallback to DLQ."""
try:
self.process_record(record)
except MalformedRecordError as e:
logger.error(f"Malformed record: {e}")
# Send to DLQ for manual review
self.db.execute("""
INSERT INTO dead_letter_queue
(original_record, error, timestamp)
VALUES (%s, %s, %s)
""", [record, str(e), now()])
# Don't raise; continue processing
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise # Stop the pipeline on unexpected errors
Now you have two queues:
- Main pipeline: Good records that processed successfully
- DLQ: Bad records that need manual review
Your pipeline never gets blocked by malformed data. You handle it asynchronously.
Putting It Together
A production data pipeline looks like this:
class ProductionPipeline:
def __init__(self, db, pipeline_name):
self.db = db
self.name = pipeline_name
self.checkpoint = CheckpointedPipeline(db, pipeline_name)
self.idempotent = IdempotentPipeline(db)
self.resilient = ResilientPipeline(db)
self.observability = ObservablePipeline(db, pipeline_name)
def run(self):
"""Main pipeline loop."""
records = self.fetch_source_data()
for record in records:
try:
# Idempotent processing
result = self.idempotent.process_record(record)
# Resilient enrichment
enriched = self.resilient.enrich_record(result)
# Store
self.db.insert('processed_data', enriched)
# Checkpoint
self.checkpoint.save_checkpoint(record['id'])
except MalformedRecordError as e:
# DLQ
self.send_to_dlq(record, e)
except Exception as e:
# Unexpected—stop and alert
logger.error(f"Pipeline failed: {e}")
raise
# Monitor
lag = self.observability.get_pipeline_lag()
print(f"Pipeline lag: {lag}s")
# Run it
pipeline = ProductionPipeline(db, "claims_processing")
pipeline.run()
This pipeline:
- ✅ Resumes from checkpoints if it crashes
- ✅ Handles duplicates safely (idempotency)
- ✅ Keeps processing even if external APIs fail
- ✅ Routes malformed data to DLQ
- ✅ Emits metrics so you know what's happening
- ✅ Fails fast on unexpected errors
The 2 AM Test
The real test of a data pipeline is whether it survives the 2 AM failure. Your database goes down for 10 minutes. Your external API returns 500 errors. A corrupted record enters the pipeline.
With these patterns:
- The pipeline continues processing (where possible)
- Failures are logged and visible
- When services recover, the pipeline resumes
- You don't wake up to "data is missing"
That's the goal.
Building data pipelines? We've helped enterprises move from "we hope it works" to "we know exactly what's happening." Let's talk about your architecture.