--- name: phy-pipeline-contract-enforcer description: Data pipeline contract enforcer. Define the expected schema at each pipeline stage boundary — field names, types, nullability, value ranges, business invariants — and validate actual data samples against those contracts. Catches schema drift between pipeline stages before it reaches production. Supports dbt models, Spark DataFrames, Pandas DataFrames, Kafka topics, REST API payloads, and raw CSV/JSON files. Can auto-generate a contract from a sample, validate a sample against an existing contract, detect when a contract has been broken by upstream changes, and produce a migration plan to fix violations. Zero external API — pure local file and CLI analysis. Triggers on "pipeline contract", "schema drift", "validate pipeline output", "data contract", "pipeline schema mismatch", "contract enforcement", "/pipeline-contract". license: Apache-2.0 metadata: author: PHY041 version: "1.0.0" tags: - data-engineering - data-quality - pipeline - schema-validation - dbt - data-contracts - great-expectations - data-mesh - developer-tools - testing --- # Pipeline Contract Enforcer Data pipelines fail silently. A column goes nullable in staging. An upstream team renames a field. A new data source adds an unexpected type. By the time it surfaces in a dashboard or an alert, the corrupt data is already in production. Paste a data sample and a contract (or generate the contract from the sample), and get an instant audit: which fields violate the contract, what the violation is, and what to fix. **Works with dbt, Spark, Pandas, Kafka, REST APIs, CSV/JSON. Zero external services. No Great Expectations config required.** --- ## Trigger Phrases - "pipeline contract", "data contract", "schema enforcement" - "validate my pipeline output", "schema drift", "pipeline schema changed" - "my downstream job broke", "column type mismatch", "unexpected null" - "generate contract from sample", "write a data contract" - "dbt schema contract", "validate DataFrame schema" - "/pipeline-contract" --- ## How to Provide Input ```bash # Option 1: Generate a contract from a sample file /pipeline-contract --generate sample.json /pipeline-contract --generate output.csv # Option 2: Validate a sample against an existing contract /pipeline-contract --validate sample.json --contract contracts/orders.yaml # Option 3: Check for drift between two samples (old vs new) /pipeline-contract --diff old-output.json new-output.json # Option 4: From dbt schema.yml /pipeline-contract --from-dbt models/schema.yml --validate data/orders.csv # Option 5: Validate a Pandas/Spark DataFrame (paste the schema) /pipeline-contract --dataframe [paste df.dtypes or df.schema output here] # Option 6: Full pipeline audit (multiple stage files) /pipeline-contract --audit pipeline/ # Reads: pipeline/stage-1.json, pipeline/stage-2.json, contracts/*.yaml ``` --- ## Step 1: Infer Pipeline Stage Type Identify what kind of pipeline artifact is being validated: ```bash # Detect file type and format file_ext="${1##*.}" case "$file_ext" in json) STAGE_TYPE="json_payload" ;; csv) STAGE_TYPE="tabular_csv" ;; parquet) STAGE_TYPE="parquet" ;; yaml|yml) STAGE_TYPE="contract_or_dbt" ;; *) STAGE_TYPE="unknown" ;; esac # Check if it looks like dbt schema.yml if grep -q "^models:" "$1" 2>/dev/null || grep -q "^version:" "$1" 2>/dev/null; then STAGE_TYPE="dbt_schema" fi # Check if it's a Python type annotation block (DataFrame schema) if echo "$1" | grep -qE "dtype|object|int64|float64|datetime64"; then STAGE_TYPE="pandas_schema" fi ``` ### Supported Stage Types | Stage Type | Detection | Examples | |-----------|-----------|---------| | JSON payload | `.json` extension or curly braces | REST API output, Kafka message | | Tabular CSV | `.csv` extension | ETL output, export files | | dbt schema | `models:` / `version:` key | dbt `schema.yml` | | Pandas schema | `dtype`/`int64`/`float64` patterns | `df.dtypes` output | | Spark schema | `StructType` / `StructField` | `df.printSchema()` output | | Parquet | `.parquet` extension | Processed pipeline files | --- ## Step 2: Extract Schema from Sample From any data sample, extract the implicit schema: ### JSON / REST Payload ```python # For each field in the JSON object, infer: # - field name # - data type (string, integer, float, boolean, null, array, object) # - nullable (does the field ever appear as null or absent?) # - example values (first 3 distinct values) # - cardinality hint (if < 20 distinct values across samples → likely enum) def infer_json_schema(samples: list[dict]) -> dict: schema = {} for sample in samples: for key, value in sample.items(): if key not in schema: schema[key] = { "type": type(value).__name__, "nullable": False, "examples": [], "values_seen": set() } if value is None: schema[key]["nullable"] = True schema[key]["values_seen"].add(str(value)[:50]) if len(schema[key]["examples"]) < 3: schema[key]["examples"].append(value) # Detect enums: < 20 distinct values for field in schema.values(): if len(field["values_seen"]) < 20: field["enum_hint"] = sorted(field["values_seen"]) return schema ``` ### CSV / Tabular ```bash # Get column names, types, null counts, sample values python3 -c " import csv, sys from collections import defaultdict, Counter with open('$FILE') as f: reader = csv.DictReader(f) rows = list(reader) schema = defaultdict(lambda: {'types': Counter(), 'null_count': 0, 'examples': []}) for row in rows: for col, val in row.items(): schema[col]['null_count'] += 1 if not val.strip() else 0 schema[col]['types'][type(val).__name__] += 1 if len(schema[col]['examples']) < 3: schema[col]['examples'].append(val) for col, info in schema.items(): null_pct = round(100 * info['null_count'] / len(rows), 1) print(f'{col}: {dict(info[\"types\"])} | null={null_pct}% | examples={info[\"examples\"]}') " ``` ### dbt schema.yml ```bash # Extract column contracts already defined grep -A 20 "columns:" "$FILE" | grep -E "name:|description:|tests:" | head -50 ``` --- ## Step 3: Contract Format A Pipeline Contract is a YAML file that defines the expected schema at a specific stage boundary: ```yaml # contracts/orders-output.yaml contract: name: orders-output version: "1.0" description: "Expected schema at the output of the orders processing stage" stage: orders_processor produces: downstream-billing, downstream-reporting fields: order_id: type: string nullable: false pattern: "^ORD-[0-9]{8}$" # Regex pattern check unique: true customer_id: type: string nullable: false order_total: type: float nullable: false min: 0.01 max: 999999.99 status: type: string nullable: false enum: [pending, confirmed, shipped, delivered, cancelled] created_at: type: datetime nullable: false format: "ISO 8601" # 2026-03-18T10:42:00Z discount_code: type: string nullable: true # Explicitly allowed null metadata: type: object nullable: true required_keys: [source, version] # If not null, these keys must exist invariants: - name: "order_total matches items" description: "order_total must equal sum of line_items[*].price" severity: error - name: "shipped orders have tracking" description: "If status=shipped, tracking_number must not be null" severity: error - name: "discount positive" description: "discount_amount must be 0 or positive" severity: warning row_count: min: 1 max: null # No upper bound ``` --- ## Step 4: Validate Sample Against Contract For each field in the contract, check the sample data: ```python def validate_against_contract(data: list[dict], contract: dict) -> list[dict]: violations = [] for row_num, row in enumerate(data): for field_name, rules in contract["fields"].items(): # CHECK: Required field present if field_name not in row and not rules.get("nullable"): violations.append({ "row": row_num, "field": field_name, "violation": "MISSING_REQUIRED_FIELD", "expected": f"field '{field_name}' must be present", "got": "field absent" }) continue value = row.get(field_name) # CHECK: Null constraint if value is None and not rules.get("nullable"): violations.append({ "row": row_num, "field": field_name, "violation": "UNEXPECTED_NULL", "expected": "not null", "got": "null" }) continue # CHECK: Type if value is not None: expected_type = rules.get("type") actual_type = type(value).__name__ if not type_matches(actual_type, expected_type): violations.append({ "row": row_num, "field": field_name, "violation": "TYPE_MISMATCH", "expected": expected_type, "got": actual_type }) # CHECK: Enum if "enum" in rules and value not in rules["enum"] and value is not None: violations.append({ "row": row_num, "field": field_name, "violation": "INVALID_ENUM_VALUE", "expected": f"one of {rules['enum']}", "got": value }) # CHECK: Range if "min" in rules and isinstance(value, (int, float)) and value < rules["min"]: violations.append({ "row": row_num, "field": field_name, "violation": "BELOW_MINIMUM", "expected": f">= {rules['min']}", "got": value }) # CHECK: Pattern import re if "pattern" in rules and value is not None: if not re.match(rules["pattern"], str(value)): violations.append({ "row": row_num, "field": field_name, "violation": "PATTERN_MISMATCH", "expected": rules["pattern"], "got": value }) return violations ``` --- ## Step 5: Drift Detection (Old vs New Samples) When comparing two versions of a pipeline output: ```bash # Extract field lists from both samples python3 -c " import json, sys old = json.load(open('old-output.json')) new = json.load(open('new-output.json')) # Normalize to list of records old_records = old if isinstance(old, list) else [old] new_records = new if isinstance(new, list) else [new] old_fields = set(old_records[0].keys()) if old_records else set() new_fields = set(new_records[0].keys()) if new_records else set() added = new_fields - old_fields removed = old_fields - new_fields common = old_fields & new_fields print('ADDED:', sorted(added)) print('REMOVED:', sorted(removed)) # Check type changes on common fields for field in sorted(common): old_type = type(old_records[0].get(field)).__name__ new_type = type(new_records[0].get(field)).__name__ if old_type != new_type: print(f'TYPE CHANGED: {field}: {old_type} -> {new_type}') # Check nullable change old_null = any(r.get(field) is None for r in old_records) new_null = any(r.get(field) is None for r in new_records) if not old_null and new_null: print(f'BECAME NULLABLE: {field}') if old_null and not new_null: print(f'NOW NON-NULL: {field} (stricter)') " ``` --- ## Step 6: Output Report ```markdown ## Pipeline Contract Report Stage: orders_processor | Contract: contracts/orders-output.yaml Sample: data/orders-2026-03-18.json (1,247 rows) ### Summary | Category | Count | Severity | |----------|-------|----------| | 🔴 Type mismatch | 3 | Error — downstream will crash | | 🔴 Unexpected null in required field | 8 | Error — violates contract | | 🟠 Invalid enum value | 12 | Error — invalid state | | 🟡 Pattern mismatch | 2 | Warning — format inconsistency | | ✅ Fields passing all checks | 14 / 18 | — | Contract status: **BROKEN** — fix before promoting to next stage --- ### 🔴 ERRORS — Pipeline Cannot Proceed Safely **TYPE MISMATCH: `order_total` (rows 42, 118, 891)** - Contract: `float` - Got: `string` (`"49.99"`, `"120.00"`, `"7.50"`) - Root cause: upstream CSV export is wrapping numbers in quotes - Fix: Add `df['order_total'] = pd.to_numeric(df['order_total'])` before output - Impact: Downstream billing service expects float — will throw TypeError on those 3 rows **UNEXPECTED NULL: `customer_id` (8 rows)** - Row indices: 15, 77, 203, 344, 502, 601, 788, 934 - Contract: `nullable: false` - Got: `null` - Root cause: Guest checkout orders — customer_id not assigned for non-registered users - Options: 1. Change contract to `nullable: true` if guest checkout is intentional 2. Assign a synthetic `guest_XXXX` ID in the upstream transformer 3. Filter out guest orders before this pipeline stage **INVALID ENUM VALUE: `status` field (12 rows)** - Values found: `"processing"` (8 rows), `"refunded"` (4 rows) - Contract enum: `[pending, confirmed, shipped, delivered, cancelled]` - Root cause: New statuses added to order system without updating pipeline contract - Fix: Update contract to add `processing` and `refunded`, or map them to existing statuses in transformer --- ### 🟡 WARNINGS **PATTERN MISMATCH: `order_id` (2 rows)** - Contract pattern: `^ORD-[0-9]{8}$` - Got: `"ORD-123"` (row 88), `"ORD-9999-X"` (row 445) - Likely test/legacy records — filter before production pipeline --- ### Drift Report (vs previous run 2026-03-17) | Change | Field | Impact | |--------|-------|--------| | ➕ NEW FIELD | `refund_amount` | New field — add to contract | | 🔄 TYPE CHANGED | `order_total` string → string (was float) | Breaking — see error above | | 🔄 BECAME NULLABLE | `discount_code` | Was non-null, now null allowed — update contract or fix upstream | | ➖ REMOVED FIELD | `legacy_channel` | Was in yesterday's output, missing today — check if intentional | --- ### Generated Contract (from this sample) Based on the actual data, here is the inferred contract to use as a starting point: ```yaml contract: name: orders-output version: "1.0" stage: orders_processor fields: order_id: type: string nullable: false pattern: "^ORD-[0-9]{8}$" customer_id: type: string nullable: true # 8 nulls observed — confirm intent order_total: type: float nullable: false min: 0.01 status: type: string nullable: false enum: [pending, confirmed, shipped, delivered, cancelled, processing, refunded] created_at: type: string # datetime — add format: ISO8601 after confirming format nullable: false discount_code: type: string nullable: true refund_amount: type: float nullable: true # New field — confirm semantics ``` --- ### Fix Priority ``` 1. Fix order_total string → float conversion in transformer (3 rows, crash risk) 2. Decide: guest checkout nulls → contract update or upstream fix (8 rows, silent corruption) 3. Add 'processing' and 'refunded' to status enum (12 rows, data loss risk) 4. Investigate legacy_channel removal (breaking change for downstream consumers?) 5. Update contract: add refund_amount field ``` --- ### dbt Integration If using dbt, add these tests to `models/schema.yml`: ```yaml models: - name: orders_output columns: - name: order_id tests: - not_null - unique - name: customer_id tests: - not_null # Remove if guest checkout is valid - name: order_total tests: - not_null - name: status tests: - accepted_values: values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled'] ``` ``` --- ## Quick Mode Fast one-line summary for CI/CD checks: ``` Pipeline Contract: orders_processor → BROKEN 🔴 3 errors (type mismatch: order_total, unexpected nulls: customer_id, invalid enum: status) 🟡 2 warnings (pattern mismatch: order_id) 📈 1 new field (refund_amount) not in contract Promote to next stage? NO — fix 3 errors first. /pipeline-contract --validate for full report with row numbers and fix instructions ``` --- ## CI/CD Integration Use in automated pipelines to gate stage promotion: ```bash # In a CI step after each pipeline stage runs: # 1. Capture the output sample head -n 100 pipeline/stage-2-output.json > /tmp/sample.json # 2. Ask Claude to validate # /pipeline-contract --validate /tmp/sample.json --contract contracts/stage-2.yaml # 3. Exit 1 if contract is BROKEN to fail the CI check # Claude will output: CONTRACT_STATUS=BROKEN or CONTRACT_STATUS=PASSING ``` --- ## Comparison to Existing Tools | Tool | Approach | When to use | |------|---------|------------| | **Great Expectations** | Python library, full test suite | Production pipelines with engineering team | | **dbt tests** | YAML tests in dbt project | dbt-specific models only | | **Soda Core** | Managed service, YAML scans | Teams with budget for managed data quality | | **This skill** | Agent-native, zero config | One-off audits, local dev, design-time contract authoring, cross-team communication | This skill fills the gap **before** you invest in a full data quality platform: understand your data's shape, write the contract, validate it locally. When you're ready to automate at scale, the contract YAML you generate here can be translated directly to Great Expectations or dbt tests. --- ## Why Contracts at Stage Boundaries The classic pipeline failure mode: ``` Stage A outputs: { "price": "12.99" } ← string Stage B expects: { "price": 12.99 } ← float Stage B runs. No error thrown. Aggregation: SUM("12.99" + "8.50") = "12.998.50" (string concat) Dashboard shows: Revenue = $1.3M (should be $0.21M) Discovery: 3 weeks later, in a board meeting. ``` A contract at the Stage A → Stage B boundary would have caught this in seconds.