Overview
Coordinate end-to-end analytical pipelines with checkpoint/resume capabilities and activity-based depth adjustment. The core principle: every stage that can fail must be resumable, and every pipeline must adapt its depth to the data it encounters.
When to Use
- Pipeline processes multiple stages sequentially and failures lose hours of work
- Large data corpus requires chunked processing with rate-limit awareness
- Analysis depth should scale with data volume (shallow scan vs. deep dive)
- Long-running jobs need graceful interruption and restart
- Multiple data sources must be combined across stages with partial-failure tolerance
When NOT to use:
- Single-shot scripts under 5 minutes with no intermediate state
- Pipelines where full re-execution is cheap and acceptable
- Real-time streaming (use event-driven patterns instead)
Core Pattern
WITHOUT orchestration: WITH orchestration:
fetch_all_data() --> Stage 1: fetch [checkpoint] -->
process_everything() --> Stage 2: parse [checkpoint] -->
analyze_all() --> Stage 3: enrich [checkpoint] -->
generate_report() Stage 4: report [checkpoint]
Fails at analyze? Fails at enrich?
Start over from scratch. Resume from Stage 3 checkpoint.
Quick Reference
| Concept | Pattern | Anti-pattern |
|---|---|---|
| Checkpointing | JSON/pickle per stage, atomic writes | No checkpoints on jobs > 5 min |
| Idempotency | Delete-then-write, upsert by key | Append-only without dedup |
| Depth adjustment | Scale analysis tiers to item count | Hardcoded depth for all inputs |
| Rate limits | Exponential backoff with jitter | Retry loops without delay |
| Progress | Structured log per stage with counts | Silent processing, no visibility |
| Failure handling | Isolate stages, continue on partial | Abort entire pipeline on one error |
Pipeline Architecture
1. Define stages as idempotent units
Each stage reads an input artifact, produces an output artifact, and writes a checkpoint on success. Re-running a completed stage is a no-op.
import json, os, time, hashlib
from pathlib import Path
class PipelineStage:
"""Base for an idempotent, checkpointed pipeline stage."""
def __init__(self, name, checkpoint_dir="checkpoints"):
self.name = name
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
@property
def checkpoint_path(self):
return self.checkpoint_dir / f"{self.name}.json"
def is_complete(self):
"""Check if this stage already finished successfully."""
if not self.checkpoint_path.exists():
return False
cp = json.loads(self.checkpoint_path.read_text())
return cp.get("status") == "complete"
def load_checkpoint(self):
"""Load partial progress for resume."""
if self.checkpoint_path.exists():
return json.loads(self.checkpoint_path.read_text())
return {"status": "not_started", "processed": [], "errors": []}
def save_checkpoint(self, state):
"""Atomic checkpoint write: write tmp then rename."""
tmp = self.checkpoint_path.with_suffix(".tmp")
tmp.write_text(json.dumps(state, indent=2, default=str))
tmp.rename(self.checkpoint_path) # atomic on POSIX
def run(self, input_data):
"""Execute stage with checkpoint/resume logic."""
if self.is_complete():
print(f"[{self.name}] Already complete, skipping.")
return self._load_output()
state = self.load_checkpoint()
already_done = set(state.get("processed", []))
items = [i for i in input_data if self._item_key(i) not in already_done]
print(f"[{self.name}] Processing {len(items)} items "
f"({len(already_done)} already done)")
for i, item in enumerate(items):
try:
result = self.process_item(item)
state["processed"].append(self._item_key(item))
# Checkpoint every N items for large batches
if (i + 1) % self.checkpoint_interval == 0:
self.save_checkpoint(state)
except Exception as e:
state["errors"].append({
"item": self._item_key(item),
"error": str(e), "time": time.time()
})
self.save_checkpoint(state)
# Continue processing; don't abort the whole stage
continue
state["status"] = "complete"
state["completed_at"] = time.time()
self.save_checkpoint(state)
return self._load_output()
# Override these in subclasses
def process_item(self, item):
raise NotImplementedError
def _item_key(self, item):
"""Unique key for deduplication."""
return hashlib.md5(json.dumps(item, sort_keys=True).encode()).hexdigest()
def _load_output(self):
"""Load this stage's output artifact."""
raise NotImplementedError
checkpoint_interval = 50 # save checkpoint every N items
2. Orchestrate stages with a pipeline runner
class Pipeline:
"""Run stages in sequence with checkpoint awareness."""
def __init__(self, stages, checkpoint_dir="checkpoints"):
self.stages = stages
self.checkpoint_dir = Path(checkpoint_dir)
self.log = []
def run(self, initial_input):
data = initial_input
for stage in self.stages:
t0 = time.time()
try:
data = stage.run(data)
elapsed = time.time() - t0
self.log.append({
"stage": stage.name, "status": "complete",
"elapsed_s": round(elapsed, 1),
"items": len(stage.load_checkpoint().get("processed", [])),
"errors": len(stage.load_checkpoint().get("errors", []))
})
except KeyboardInterrupt:
self.log.append({"stage": stage.name, "status": "interrupted"})
print(f"\n[pipeline] Interrupted at {stage.name}. "
f"Re-run to resume from this stage.")
break
except Exception as e:
self.log.append({
"stage": stage.name, "status": "failed", "error": str(e)
})
print(f"[pipeline] Stage {stage.name} failed: {e}")
break
return data
def write_report(self, path="docs/analysis/03-automated-orchestration.md"):
"""Write pipeline status report."""
report = self._build_report()
Path(path).parent.mkdir(parents=True, exist_ok=True)
Path(path).write_text(report)
def _build_report(self):
lines = ["# Automated Orchestration Report\n"]
lines.append(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
lines.append("## Pipeline Status\n")
lines.append("| Stage | Status | Items | Errors | Time (s) |")
lines.append("|-------|--------|-------|--------|----------|")
for entry in self.log:
lines.append(
f"| {entry.get('stage','')} "
f"| {entry.get('status','')} "
f"| {entry.get('items','-')} "
f"| {entry.get('errors','-')} "
f"| {entry.get('elapsed_s','-')} |"
)
# Summarize interruptions / skips
failures = [e for e in self.log if e["status"] != "complete"]
if failures:
lines.append("\n## Interruptions and Failures\n")
for f in failures:
lines.append(f"- **{f['stage']}**: {f['status']}"
+ (f" -- {f['error']}" if "error" in f else ""))
else:
lines.append("\n## Result\n")
lines.append("All stages completed successfully.")
return "\n".join(lines) + "\n"
3. Activity-based depth adjustment
Scale analysis intensity to data volume. Avoid spending deep-analysis resources on tiny datasets, or shallow-scanning massive ones without prioritization.
def choose_depth(item_count, thresholds=None):
"""Select analysis depth tier based on item count.
Returns a dict of parameters that downstream stages consume.
Callers should NOT hardcode depth -- always derive from data.
"""
thresholds = thresholds or {
"shallow": (0, 100), # quick summary only
"standard": (101, 5000), # full analysis per item
"deep": (5001, 50000), # sampled deep + full summary
"archival": (50001, None), # statistical sampling + key items only
}
for tier, (lo, hi) in thresholds.items():
if hi is None and item_count >= lo:
return {"tier": tier, "sample_rate": 0.05, "full_analysis": False}
if lo <= item_count <= hi:
return {"tier": tier,
"sample_rate": 1.0 if tier in ("shallow","standard") else 0.2,
"full_analysis": tier in ("shallow", "standard")}
return {"tier": "standard", "sample_rate": 1.0, "full_analysis": True}
4. Rate-limit-aware fetching
import random
def fetch_with_backoff(fetch_fn, max_retries=5, base_delay=1.0):
"""Retry with exponential backoff + jitter. Respects rate limits."""
for attempt in range(max_retries):
try:
return fetch_fn()
except RateLimitError:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f" Rate limited. Waiting {delay:.1f}s (attempt {attempt+1})")
time.sleep(delay)
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(base_delay)
raise RuntimeError(f"Failed after {max_retries} retries")
Handling Insufficient Data
| Situation | Action |
|---|---|
| Pipeline has nothing to process | Log "no input data", write report noting empty input, exit cleanly with status "no_data" |
| A stage produces empty output | Log warning, pass empty collection to next stage, do NOT skip or error. Let downstream stages handle empty input gracefully |
| External API unavailable | Retry with backoff. After max retries, mark stage as "degraded", continue with cached/partial data if available. Log what was skipped |
| Checkpoint file corrupted | Delete checkpoint, re-run stage from scratch. Log the corruption event |
Common Mistakes
| Mistake | Fix |
|---|---|
| No checkpointing on jobs > 5 min | Add checkpoint writes every N items and on stage completion |
| Monolithic single-function pipeline | Split into discrete stages with clear input/output contracts |
| Silently skipping failed items | Log every failure with item ID, error, timestamp |
| Hardcoded analysis depth | Use choose_depth() or equivalent based on actual item count |
| Retry without backoff | Always exponential backoff with jitter for external calls |
| Non-atomic checkpoint writes | Write to .tmp then rename (atomic on POSIX) |
| Ignoring partial failures | Collect errors, continue processing, summarize in report |
| No progress visibility | Log stage entry/exit, item counts, elapsed time |
Boundaries
SHOULD do:
- Manage pipeline stage sequencing and dependencies
- Handle interruptions gracefully (Ctrl+C saves checkpoint)
- Scale depth to data size automatically
- Log progress with structured, parseable output
- Produce a status report at
docs/analysis/03-automated-orchestration.md
Should NOT do:
- Make analytical conclusions (that is for domain-specific downstream skills)
- Silently skip failed stages without logging
- Retry indefinitely without backoff
- Decide what data means -- only move it through stages reliably
Report Output
The pipeline MUST end by writing its status to docs/analysis/03-automated-orchestration.md containing:
- Pipeline run timestamp
- Table of stages: name, status (complete/failed/interrupted/skipped), item count, error count, elapsed time
- Section listing any interruptions or skipped steps with reasons
- Summary of depth tier selected and why