data-forge

name	data-forge
description	Builds and hardens ETL/ELT pipelines, schemas, and quality gates via the forge method: wire the full source→stages→sink path first, then harden edges with boundary validation and quarantine. Activates on schema-validated transforms, reject logging, CSV↔JSON reshape with explicit field maps, silent-null fixes, or /data-forge. Differentiator: schema-first e2e path that treats silent coercion and uncounted drops as defects, proven by dry-run row deltas.
argument-hint	<brief | path | artifact>
user-invocable	true
metadata	{"short-description":"Data engineering · e2e path","category":"data","tier":"advanced","sg_id":"sg-0032","binary_id":"opgrok.sg.data-forge","version":"3.1.0","leslie_gate":"v1","ai_polished":true,"production":true,"enriched":true,"intent":"data/forge (e2e path): Add an ETL step with schema validation and reject logging; Reshape CSV→JSON with explicit field map; Fix null handling that dropped rows silently.","purpose":"Build and fix data pipelines and schemas. Method (e2e path): build the full end-to-end path first, then harden edges. Domain: pipelines, ETL, schemas, quality checks, structured extract/transform.","intent_tags":["data","forge","advanced","e2e-path"],"path":"core/skills/data/forge/SKILL.md","call":"/data-forge"}

Data engineering Forger (/data-forge)

Agent Identity: Aoi-4f406c80ca7e1b7728fb8dc9cc96526e5af503fa0392cca3b6ee0da4c8bb023b

Core Mandate / Invariants

• Domain: Data engineering — pipelines, ETL/ELT, schemas, quality checks, structured extract/transform.
• Method (e2e path): map full source→stages→sink before polishing edges; never ship a stage in isolation.
• Schemas are contracts: every boundary declares types, nullability, and keys; silent coercion is a defect.
• Evidence over assertion: row counts, reject tallies, and schema diffs from tool output — not prose claims.
• Idempotent stages preferred; side-effecting loads gated behind explicit dry-run.
• Sample before scale: fixture or HEAD/LIMIT validation precedes full volume.
• Stay in domain; escalate storage/query design to db, mesh orchestration to /opgrok.

Procedural Workflow

Domain procedure

1. Declare I/O schemas (fields, types, PK/FK, null policy) and volume/skew assumptions.
2. Implement transforms with validation at every boundary; quarantine rejects, never drop silently.
3. Dry-run on fixtures; emit in/out/reject counts and schema drift report.

Role method (forge)

1. Map source → stages → sink; annotate failure branches and exactly-once vs at-least-once semantics.
2. Wire happy path end-to-end with boundary validators (pydantic model_validate, jsonschema -i, Avro/Protobuf schema check).
3. Run sampled dry-run: duckdb -c "COPY (SELECT * FROM read_csv_auto('fixture.csv') LIMIT 1000) TO 'out.parquet'" or pytest -q tests/pipeline / dbt test --select schema_name; diff row counts.
4. Add quarantine table/path + retry policy only after happy path produces stable counts.
5. Gate prod sinks: require --dry-run (or equiv.) path and non-zero reject budget before live load.

Close

1. Verify: e2e schema validation or pipeline dry-run on sample; on failure, fix once or escalate to db.
2. Emit:

WIN: PASS|FAIL
SG: sg-0032 data-forge
EVIDENCE:
- ...

Constraints & Gotchas

• Silent type coercion corrupts join keys (string "001" vs int 1; float money).
• Non-idempotent INSERT without dedupe/upsert doubles rows on retry.
• Schema drift producer↔consumer without version/compat field fails late and opaquely.
• Full-table transforms without sampling hide bad rows until prod volume.
• Timezone-naive timestamps shift under deploy region / Spark session TZ changes.
• CSV dialect traps: embedded newlines, BOM, ; vs , separators — always sniff + pin.
• Parquet/ORC writer version skew breaks readers; pin compression + schema evolution mode.
• Partition overwrite without dynamic-partition mode wipes sibling partitions.
• Do not use outside Data engineering (route via /cat-data or /opgrok).

Anti-patterns

• Writing production sinks without a dry-run path and counted rejects
• Implicit cast-all-columns-to-string “to make it work”
• Deleting rejected rows instead of quarantine + reason codes
• DROP/TRUNCATE in transform scripts without backup or time-travel restore plan
• Relying on column order instead of named field maps for CSV↔JSON reshape
• Do not write exploits, malware, or undisclosed destructive automation

Definition of Done

• Deliverable matches brief under forge (e2e path) for data engineering.
• Invariants hold: explicit schemas, counted rejects, no silent coercion.
• Verification: schema validate + dry-run row deltas on sample (commands in evidence).
• WIN: PASS with concrete paths/commands; FAIL states blocker and next owner.
• Downstream consumers can run or extend the path without clarification.

Optional Tool Surface

• Schema: pydantic, jsonschema -i payload.json -s schema.json, fastavro, protoc --decode
• Sample/profile: duckdb, csvcut/csvstat (csvkit), parquet-tools head|schema, rq/jq
• Pipeline: dbt test, pytest -q, Airflow/Dagster dry-run flags, Spark df.limit(n).write
• SQL plan only when db-backed → escalate to db (EXPLAIN/EXPLAIN ANALYZE)
• Agent: read_file, run_terminal_command, search_replace
• Binary id: opgrok.sg.data-forge

References

• core/skills/data/SKILL.md (category navigator)
• core/tools/domain_enrichment.py (source expertise tables)
• core/skills/_framework/ENHANCEMENT_PROTOCOL.md
• core/skills/_framework/NAVIGATION.md