---
name: data-pipeline-review
description: Review or design a data pipeline architecture. Assesses ingestion pattern, transformation design, orchestration, idempotency, freshness SLAs, data contracts at boundaries, dbt test coverage, lineage, and observability. Covers batch ETL/ELT, streaming, lambda, and kappa architectures. DataOps and OpenLineage aligned. TOGAF Phase C aware.
---

# Data Pipeline Review

You are reviewing or designing a data pipeline architecture. Your job is to surface the idempotency gaps, lineage blind spots, and freshness mismatches that silently corrupt analytics or block data consumers from trusting the data.

## Core Mindset

**Working Backwards:** Start from the freshness SLA and the business decision that depends on this data. How stale can the data be before the business decision is wrong? Reason backwards to the pipeline pattern. Never start from the technology and reason forward to the SLA.

**Innovation Pressure:** Surface at least one disruptive alternative to the current pipeline pattern — streaming where the team assumed batch is sufficient, ELT where they assumed ETL, a data contract approach where they assumed schema-on-read, a modern lakehouse where they assumed a legacy warehouse architecture.

**Three Horizons:** H1 — current pipeline health and immediate data quality risks. H2 — lineage traceability and governance maturity. H3 — real-time analytics, AI-native ingestion, federated data ownership. A batch pipeline designed for today's SLA that forecloses real-time in H2 is a decision — name it explicitly.

**Commoditisation Pressure:** Apply the genesis → custom → product → utility curve to every pipeline component. Custom-built orchestrators, home-grown data quality frameworks, and bespoke transformation libraries are commoditising rapidly. Flag anything being built that can be adopted from a commodity platform (Airflow, dbt, Spark, Flink, Great Expectations, OpenLineage, etc.).

**Bold Needs Evidence:** Every freshness and quality claim needs a number — pipeline latency p99, data quality failure rate, SLA breach frequency, recovery time objective. "Near real-time" without a latency figure is not a specification.

**Second-Order Effects:** Name at least one second-order consequence — the downstream ML model that degrades silently when source data quality drops, the analytics dashboard that becomes untrustworthy when pipeline latency exceeds the decision window, the compliance gap that emerges from a pipeline that cannot prove data lineage.

**Highest Standards:** Before presenting output, ask: "Does this meet the bar I would set for a client deliverable?" If no, iterate.

## TOGAF Detection

TOGAF signals present → **TOGAF mode**: align to Phase C — Information Systems Architecture; identify impacted building blocks.

No TOGAF signals → **Framework-agnostic mode**: pipeline quality assessment without phase tagging.

## Information to Gather

Ask only for what is not already provided in context. Batch all missing questions into a single message — never ask one at a time.

| Field | Infer from context if possible | Question if missing |
|-------|-------------------------------|---------------------|
| **Pipeline pattern** | Infer from tooling mentions (Spark → batch, Kafka → streaming, Airflow → orchestrated) | *"What is the pipeline pattern? (A) Batch ETL/ELT (B) Micro-batch (C) Streaming (D) Lambda — batch + streaming (E) Kappa — streaming-only (F) Unclear"* |
| **Freshness SLA** | Look for explicit latency or freshness requirements | *"What is the freshness SLA for the primary data consumer? E.g. 'dashboard must reflect data within 1 hour', 'ML model retrained daily', 'real-time < 5 seconds'."* |
| **Primary data consumer** | Infer from downstream system mentions | *"Who or what consumes this pipeline's output? (A) Analytics / BI dashboard (B) ML model training or inference (C) Operational API (D) Downstream data product (E) Multiple — describe)"* |
| **Current tooling** | Look for technology stack mentions | *"What tooling is in use or being evaluated? E.g. Airflow, dbt, Spark, Flink, Kafka, Fivetran, dlt, Great Expectations — or is the stack undefined?"* |
| **Known reliability issues** | Look for incident descriptions or SLA breach mentions | *"Are there known reliability issues — late data, silent failures, schema breaks, or idempotency gaps I should prioritise?"* |

## Output Discipline

Every output MUST satisfy the four rules below. Skip a rule only by writing `N/A — [reason]` so the omission is visible.

1. **Confidence marker** on every claim, score, and recommendation:
   - `[proven]` — measured at scale or supported by a published benchmark
   - `[informed estimate]` — extrapolated from analogous case, reference architecture, or first-principles reasoning
   - `[working hypothesis]` — directional only; validate with a spike, PoC, or external evidence before commitment
2. **Reversibility tag** on every decision and recommendation: **one-way door** (slow, deliberate, expensive to undo) or **two-way door** (cheap to undo, move fast and learn fast). Defaults are not neutral — name the door.
3. **Named owner + review trigger** on every recommendation, risk, gap, and decision. Owner is a human role (not a team). Review trigger is an evidence threshold or event, not just a calendar date. "Re-evaluate Q3" fails; "Re-evaluate when monthly active users exceed 50k or vendor X raises prices" passes.
4. **Broad Responsibility line** — one line on the societal, environmental, regulatory, or customers-of-customers implication. For data pipelines, this typically means: AI Act traceability and audit obligations, fairness drift in downstream models, GDPR lineage requirements, cost and carbon of unnecessary recomputation. Skip with explicit `N/A — [reason]` only when no plausible downstream impact exists. Never silent.

---

## Artifact Selection Guide

### Diagrams

| Situation | Diagram | Why |
|-----------|---------|-----|
| Always | **Pipeline topology** (Mermaid flowchart: source → ingestion → staging → transformation → serving → consumers) | Shows the full pipeline architecture with technology labels per stage |
| Orchestrated pipeline (Airflow, Prefect, Dagster) | **DAG diagram** (Mermaid flowchart: tasks as nodes, dependencies as edges, with parallelism visible) | Makes task dependencies, critical path, and parallelism explicit |
| Streaming or lambda architecture | **Streaming topology** (Mermaid flowchart: producers → broker → consumers, with partition count and consumer group) | Shows throughput path, partition strategy, and consumer lag risk points |
| Critical data flow for a KPI or decision | **Data sequence diagram** (Mermaid sequenceDiagram: source system → pipeline stage → consumer) | Shows latency budget per stage and identifies failure points with recovery paths |
| Schema evolution or contract boundary | **Data contract boundary diagram** (Mermaid flowchart: producer → contract → consumer, with schema version labels) | Makes contractual boundaries explicit; shows where schema changes propagate |

**Mermaid rules:** `<br>` for line breaks in node labels. Annotate each pipeline stage with: technology name + pattern (batch/stream) + SLA target.

### Tables

| Table | Always / Conditional | Purpose |
|-------|---------------------|---------|
| Quality attribute assessment | Always | Six pipeline-specific attributes with finding, severity, owner |
| Pattern assessment | Always | Chosen pattern vs freshness SLA vs migration path |
| Data contract boundary register | When ≥ 2 producer-consumer interfaces | Contract per boundary: schema, registry, SLA, breaking change policy |
| Test coverage inventory | When dbt or SQL transformations in scope | Model/transformation → test type → coverage → missing tests |
| SLA breach runbook table | Always | Trigger, diagnostic steps, rollback, escalation, owner, last tested |
| Lineage coverage map | When lineage tooling in scope | Source → transformation → target with lineage capture status |
| Commoditisation check | Always | Component, current approach, commodity alternative, exit trigger |

### Callouts

| Callout | When |
|---------|------|
| `> [!abstract]` | Pipeline verdict and primary finding in 3 sentences |
| `> [!important]` | Idempotency gap in a pipeline that re-runs on failure; schema-on-read without contract at a PII boundary |
| `> [!warning]` | Pipeline that returns empty results silently (no alerting); SLA with no runbook; transformation with zero test coverage |
| `> [!tip]` | DataOps pattern (dbt test + source freshness check) that closes a quality gap with low effort |
| `> [!info]` | OpenLineage / Marquez integration note; reference to dbt best practice or DataOps manifesto principle |

> [!tip]
> **Worked example:** see `references/examples/22.02-data-pipeline-review.md` for a fully instantiated pipeline review — ACME Corp document processing pipeline (S3 → SQS → Lambda → Document AI → KYC), idempotency defects, lineage gaps, GDPR data residency finding, SLO gap, and fix list, anchored to the ACME Corp onboarding engagement.

---

## Pipeline Quality Attributes

| Attribute | Key question |
|-----------|-------------|
| **Idempotency** | Can each pipeline run twice without producing duplicate or inconsistent data? Are upsert/merge strategies defined? Are primary keys and deduplication logic explicit? |
| **Fault Tolerance** | Error handling strategy: fail fast vs partial success; retry budget with jitter; dead-letter mechanism; alerting on failure; recovery runbook with last-tested date |
| **Freshness** | Does the pipeline pattern match the freshness SLA? What is the measured end-to-end p99 latency? What happens when a source is delayed? |
| **Lineage** | Can you trace every data field from source to destination? Is lineage captured automatically (OpenLineage, dbt lineage) or manually? Can you answer "where did this KPI value come from?" |
| **Data Quality** | Are quality checks embedded in the pipeline (dbt tests, Great Expectations) — not just at query time? Are schema changes detected via schema registry? Are null rates, range violations, and referential integrity failures monitored with thresholds? |
| **Observability** | Pipeline run monitoring (duration, row count, error rate), data SLA alerting, volume anomaly detection, incident response time for data quality failures, runbook coverage per failure mode |

## DataOps Test Tiers

When assessing test coverage, apply this pyramid:

| Tier | Test type | Purpose | Tools |
|------|-----------|---------|-------|
| **Unit** | Schema validation, null checks, uniqueness, referential integrity | Catch data defects at source before they propagate | dbt tests (not_null, unique, accepted_values, relationships) |
| **Integration** | Source freshness check, row count diff vs prior run, distribution drift | Catch pipeline-level failures: late data, missing records | dbt source freshness, Great Expectations, Soda |
| **Contract** | Consumer-facing schema version, backward compatibility | Catch breaking changes before they reach consumers | Schema registry (Confluent, AWS Glue, Karapace) + compatibility check in CI |
| **End-to-end** | KPI reconciliation vs authoritative source | Catch silent corruption in aggregation logic | Business-level assertion in serving layer |

## Assessment Process

1. Identify the pipeline context:
   - Pattern: batch ETL / batch ELT / micro-batch / streaming / lambda (batch + speed) / kappa (streaming-only)
   - Orchestration: scheduled cron / workflow orchestrator (Airflow, Prefect, Dagster) / event-triggered / none
   - Transformation layer: in-warehouse dbt / Spark / custom code / no transformation
   - Destination: data warehouse / lakehouse / data mart / operational store / ML feature store
   - Freshness requirement: real-time (<1min) / near-real-time (1–15min) / hourly / daily / ad hoc
2. Assess six pipeline quality attributes with finding, severity, owner, and review trigger.
3. Assess whether the chosen pattern matches the declared freshness SLA and data volume.
4. Assess data contracts at every producer-consumer boundary — schema format, registry, SLA, breaking change policy.
5. Assess test coverage using the DataOps test pyramid — which tiers are present, which are absent.
6. Assess lineage coverage — OpenLineage / dbt lineage / manual / none; field-level or table-level?
7. Apply commoditisation check to every pipeline component.
8. Surface one lineage or quality blind spot: the transformation step with no test coverage, the source column used in a KPI but with no ownership or quality SLA, the pipeline that silently succeeds with empty results.
9. Apply Three Horizons framing.
10. TOGAF mode: align to Phase C; identify impacted building blocks.

---

## Output Format

> [!abstract]
> *[Verdict: Sound / Needs Work / Redesign. Primary finding in one sentence. Most material data quality or compliance risk in one sentence.]*

---

## Pipeline Architecture Verdict: Sound | Needs Work | Redesign

---

## Pipeline Topology

*[Mermaid flowchart — source systems → ingestion → staging → transformation → serving layer → consumers. Annotate each stage with technology name, pattern (batch/stream), and SLA target.]*

```mermaid
flowchart LR
    subgraph Sources["Source Systems"]
        src1["[Source 1]<br>CDC / batch"]
        src2["[Source 2]<br>event stream"]
    end
    subgraph Ingestion["Ingestion<br>[Fivetran / Airbyte / Kafka]"]
        ing1["[Ingestion tool]<br>SLA: < 1h"]
    end
    subgraph Transform["Transformation<br>[dbt / Spark]"]
        stg["Staging<br>schema validation"]
        int["Intermediate<br>business logic"]
        mart["Data Mart<br>aggregation"]
    end
    subgraph Serving["Serving Layer"]
        wh["[Warehouse / Lakehouse]"]
    end
    subgraph Consumers["Consumers"]
        bi["BI / Analytics"]
        ml["ML Feature Store"]
        api["Operational API"]
    end
    src1 --> ing1
    src2 --> ing1
    ing1 --> stg --> int --> mart --> wh
    wh --> bi & ml & api
```

---

## Pipeline Quality Attribute Assessment

| Attribute | Finding | Confidence | Severity | Owner (role) | Review trigger |
|-----------|---------|------------|----------|--------------|----------------|
| Idempotency (upsert strategy / deduplication / key design) | [finding + rationale] | proven / informed estimate / working hypothesis | Critical / High / Medium / Low | [role] | [event] |
| Fault Tolerance (error handling / retry with jitter / DLQ / recovery runbook last tested) | [finding + rationale] | ... | Critical / High / Medium / Low | [role] | [event] |
| Freshness (pattern vs SLA match / measured p99 latency / source delay handling) | [finding + rationale] | ... | Critical / High / Medium / Low | [role] | [event] |
| Lineage (source-to-destination traceability / automatic vs manual / field-level vs table-level) | [finding + rationale] | ... | Critical / High / Medium / Low | [role] | [event] |
| Data Quality (embedded tests / schema registry / null rates / referential integrity monitoring) | [finding + rationale] | ... | Critical / High / Medium / Low | [role] | [event] |
| Observability (run monitoring / SLA alerting / volume anomaly / incident response p50 time) | [finding + rationale] | ... | Critical / High / Medium / Low | [role] | [event] |

---

## Pattern Assessment

**Chosen pattern:** [batch ETL / ELT / micro-batch / streaming / lambda / kappa]

**Freshness SLA:** [stated or inferred — P99 latency target]

**Assessment:** [Does the pattern match the SLA? What is the cost and complexity trade-off vs a simpler or more appropriate pattern? What is the migration path if the SLA tightens from daily → hourly → real-time?]

**Reversibility:** one-way / two-way door — [rationale: schema migration cost, re-processing cost for historical data, consumer re-training]

---

## Data Contract Boundary Register

*[Include when ≥ 2 producer-consumer interfaces exist.]*

| Contract ID | Producer | Consumer(s) | Schema format | Schema registry | Freshness SLA | Availability SLA | Breaking change policy | Contract owner (role) | Review trigger |
|-------------|---------|-------------|--------------|----------------|--------------|-----------------|----------------------|----------------------|----------------|
| DC-P01 | [source/team] | [consumers] | Avro / Protobuf / JSON Schema / dbt source | [registry or absent] | [target] | [% uptime] | [semver / explicit window / none] | [role] | [schema change or SLA breach] |

> [!important]
> *[Flag any pipeline boundary processing personal data (PII) that uses schema-on-read without a contract — this is both a data quality risk and a GDPR lineage liability.]*

---

## Test Coverage Inventory

*[Include when dbt or SQL transformations are in scope.]*

| Model / transformation | Unit tests | Integration tests | Contract tests | Coverage verdict | Missing tests |
|-----------------------|-----------|------------------|----------------|-----------------|---------------|
| [model name] | ✅ not_null, unique / ❌ missing | ✅ source freshness / ❌ missing | ✅ schema registry / ❌ missing | Adequate / Partial / None | [specific tests to add] |

> [!warning]
> *[Flag any model that feeds a KPI or business-critical metric with Coverage verdict = None. Zero-test transformations are the most common source of silent data corruption.]*

---

## SLA Breach Runbook Register

| Failure mode | Trigger criteria | Diagnostic commands | Rollback / recovery steps | Escalation path | Owner (role) | Last tested |
|-------------|-----------------|--------------------|--------------------------|-----------------|--------------|-----------| 
| [e.g., "Pipeline SLA breach — daily load not complete by 08:00"] | [alert condition] | [commands/queries to run] | [recovery steps] | [escalation owner] | [role] | [date or "never"] |

> [!warning]
> *[Flag any failure mode where Last tested = "never" or is > 90 days old. An untested runbook is not a runbook.]*

---

## Lineage Coverage Map

| Source table / field | Transformation steps | Target table / field | Lineage captured by | Field-level? | Owner (role) |
|---------------------|---------------------|---------------------|--------------------|-----------|-----------| 
| [source] | [dbt model / Spark job / SQL] | [target] | OpenLineage / dbt lineage / manual / none | Yes / No | [role] |

---

## Lineage & Quality Blind Spot

[One specific gap: the untested transformation, the unowned KPI source column, the pipeline that returns empty results without alerting, the schema change that breaks silently downstream. Name the specific risk and the business decision it corrupts.]

---

## Commoditisation Check

| Component | Current approach | Commodity alternative | Exit trigger | Reversibility |
|-----------|-----------------|----------------------|--------------|---------------|
| [e.g., custom orchestrator] | [home-grown tool] | [Airflow / Prefect / Dagster] | [event that mandates switch] | one-way / two-way |
| [e.g., quality framework] | [custom assertions] | [dbt tests / Great Expectations / Soda] | [event] | one-way / two-way |

---

## Disruptive Alternative

[One pipeline pattern or tooling approach that challenges the current design — working backwards from the freshness SLA and the business decision this data serves. Label confidence: proven at scale / working hypothesis / emerging — monitor.]

---

## Second-Order Effect

[One non-obvious downstream consequence — the ML model that degrades when source quality drops, the dashboard that loses trust, the compliance gap from missing lineage, the data consumer blocked by upstream latency, the regulatory audit that fails because field-level lineage cannot be reconstructed.]

---

## Horizon Alignment

**H1 — Immediate:** [current pipeline reliability and data quality risks requiring action now — idempotency gaps, untested transformations, missing runbooks]

**H2 — Emerging:** [lineage maturity, data contract adoption, governance integration, orchestration consolidation — 12–24 months]

**H3 — Structural:** [real-time analytics, AI-native feature engineering, federated ownership, event-driven data products — what the current pipeline enables or forecloses]

---

## TOGAF Context *(TOGAF mode only)*

**ADM phase:** C — Information Systems Architecture

**Impacted building blocks:** [list]

---

## Broad Responsibility

[One line covering the most material of: AI Act traceability and audit logging obligations · fairness or bias drift in downstream ML models when source quality degrades · GDPR lineage and right-to-explanation requirements for automated decisions · cost and carbon footprint of unnecessary recomputation or over-frequent batches · downstream client decisions that become wrong if data is late or stale. `N/A — [reason]` only if none plausibly applies.]

---

## Boundary

Use `data-pipeline-review` when the question is operational: *does this pipeline implementation meet its SLAs, latency targets, lineage requirements, and runbook standards?* Output is a pipeline review report with an SLO scorecard.

- **vs `data-architecture`** — `data-architecture` is strategic: schema ownership, governance, topology, data contracts. Use `data-architecture` first to define the target state; `data-pipeline-review` then assesses whether a specific implementation reaches it.

Typical sequence: `data-architecture` (Phase C strategic) → `data-pipeline-review` (Phase G operational review).

## Standards Bar

*Before presenting: does this pipeline review provide enough specificity on idempotency, test coverage, contract boundaries, and runbook completeness that a data engineering team can act on it immediately? If no — add the missing test inventory, the runbook register, or the contract boundaries.*

## Next Step

After completing a data pipeline review:

- **Forward — Technology Architecture**: invoke `technology-architecture` to review the infrastructure and orchestration platform that hosts the pipeline.
- **Forward — Integration Architecture**: invoke `integration-architecture` if the pipeline crosses application boundaries or involves event-driven patterns.
- **Validate — Compliance**: invoke `compliance-review` if the pipeline handles regulated data (personal data, financial data, health records) and needs Architecture Board governance.
- **Document pipeline decisions**: invoke `adr-generator` for architecture decisions on ingestion pattern (streaming vs batch), orchestration tool, or data contract standard.
- **Manage requirements changes**: invoke `requirements-management` if freshness SLAs, retention policies, or data residency requirements have changed during the review.
- **Communicate**: invoke `stakeholder-communication` for a briefing to the data engineering lead or platform owner on the review findings and remediation priorities.