Writing · contract-driven-data-pipeline

Why My Data Pipeline Rejects 84% On Purpose

Contract-Driven Data Pipeline · 11 min read · 2,478 words

A pipeline that throws away 838 of every 1,000 rows sounds broken. Mine isn't — it's doing exactly what I designed it to do, and it can name the specific rule that killed each of those 838 rows.

That second part is the whole project. Because the question every data team eventually gets asked is not "did the pipeline run?" It's "why isn't this record in the warehouse?" — and the honest answer is usually a shrug.

I built the opposite of that shrug.

The shrug is an architecture problem

The first version of this pipeline did what most pipelines do. Validation lived inline in a PySpark function: a trim() here, a regex there, a .filter() that quietly deleted anything inconvenient. A row that failed simply ceased to exist.

That's three sources of truth pretending to be one:

1. The code that accepts rows.
2. The (missing) explanation of why rows were dropped.
3. The human-readable documentation of what "valid" means — a separate Markdown file that started drifting the moment I saved it.

They disagree eventually. They always do. And when they do, nobody notices, because the evidence was deleted.

One contract, four artifacts

The fix was to pull the rules out of Spark entirely, into a pure Python module — no Spark, no I/O, no imports beyond dataclasses. Each field gets exactly one rule, expressed as data rather than logic:

FieldRule("phone", REGEX, "invalid_phone",
          "Greek mobile number (69 + 8 digits).",
          DIRECT_IDENTIFIER, pattern=PHONE_REGEX)

There are six of them — name, email, phone, zip_code, age, city — in a single frozen tuple. From that one definition I generate four things that are now structurally incapable of disagreeing:

• The accept filter. Spark compiles each rule into a column predicate and keeps only rows passing all six.
• The rejection reason. Every dropped row is tagged with the first rule it violated and quarantined to a rejects/ zone.
• The PII classification. Each field is labelled direct_identifier, quasi_identifier, or pseudonymised.
• The data dictionary. A Markdown doc rendered straight from the tuple. CI runs the generator with --check and fails the build if the committed doc has drifted.

That last one matters more than it looks. Documentation drift isn't a documentation problem — it's a governance problem with a documentation symptom. Making it a red build is the cheapest fix I know.

How Spark compiles the contract

The interesting part is that the accept path and the reject path are built from the same predicate function, so they can't diverge:

def _rule_ok(rule: FieldRule) -> Column:
    c = col(rule.field)
    if rule.kind == NON_EMPTY:
        return c.isNotNull() & (length(c) > 0)
    if rule.kind == REGEX:
        return c.isNotNull() & c.rlike(rule.pattern)
    if rule.kind == INT_RANGE:
        return c.isNotNull() & (c >= rule.minimum) & (c <= rule.maximum)

Three design choices are hiding in those six lines.

It's null-safe by construction. Every predicate starts with isNotNull(). In Spark SQL a null propagates through comparisons as null, not false — so without that guard, null rows fall out of both the accept filter and the reject filter and vanish into the gap. With it, the accepted set and the rejected set are a provable partition: every input row lands in exactly one. There's a test that asserts precisely that (test_clean_and_rejected_partition_the_input).

Both paths normalise identically. A shared _prepared() step trims the string fields and casts age to int before either path evaluates. Otherwise " Athens " could be accepted by one path and rejected by the other.

Rejection precedence is the contract's declaration order. The reject path folds the rules into a chained when(...), so a row failing three checks is attributed to the first one in the tuple. Simple, deterministic, and testable.

There are no Python UDFs anywhere — the whole transform is Spark SQL built-ins. That's not aesthetics: it means executors never need the contract module shipped to them via --py-files, and the job runs unchanged in-process or on a cluster.

The pseudonym is part of the contract too

The loaded user_id is not the source id. It's a deterministic MD5 of name || email || phone — the three fields the contract labels as direct identifiers:

df.withColumn(SURROGATE_KEY, md5(concat_ws("||", *SURROGATE_SOURCES)))

The same person always maps to the same surrogate, which is what makes the downstream ON CONFLICT (user_id) DO NOTHING upsert idempotent — you can re-run the pipeline all day without duplicating anyone. And no natural key ever becomes the join key in the warehouse.

The data dictionary explains that in prose, automatically, because both the classification and the surrogate's source fields come from the same tuple.

Secrets in Airflow, logic in plain Python

There's a quiet decision underneath all of this that makes the testing possible.

Airflow connections are declared as code — AIRFLOW_CONN_AWS_DEFAULT, AIRFLOW_CONN_POSTGRES_DEFAULT, AIRFLOW_CONN_SPARK_DEFAULT — as env vars in Compose, in JSON form (the URI form mangles AWS secret keys that contain slashes). Nobody clicks through the Airflow UI to set up a connection; docker compose up produces a working stack.

But the ETL scripts never read those connections. The DAG resolves the credentials and injects them:

client = S3Hook(aws_conn_id="aws_default").get_conn()
upload_to_s3(client, local_path, bucket, key)

The loader gets the same treatment — the task builds a connect(dbname) factory from BaseHook.get_connection("postgres_default") and hands it in. So secrets live in Airflow's connection store at orchestration time, while generate_dirty_data_S3.py and load_to_db_final.py stay ordinary Python modules that import and run without Airflow installed anywhere.

That's not a purity exercise. It's exactly why the 56-test suite and the Postgres smoke job can run in CI on a plain Python image, and why the transform can be unit-tested in a local SparkSession. The same reason the heavy imports sit inside the task functions rather than at module level: DAG parsing stays light and never breaks on a missing dependency.

Making quality visible, not assumed

Every run computes a data-quality report — total, accepted, rejected, and rejected-by-reason. That report goes to four places:

1. Logged as JSON in the Airflow task.
2. Written to data/dq_report.json.
3. Uploaded to a date-partitioned quality/dt=<ds>/ zone in S3.
4. Pushed to StatsD as airflow.dq.* gauges over UDP.

Point 4 is the one I'd fight for. Airflow already emits StatsD metrics, so the Spark task just borrows the same host and port that Airflow is configured with, and the numbers land in the exact same Prometheus → Grafana path as the operational ones. A statsd_mapping.yml rewrites the dotted metric names into properly labelled Prometheus series.

The result: task durations by stage, DAG-run duration, task finishes by state, scheduler heartbeat — and right underneath them, the accept-rate gauge, accepted-vs-rejected, and rejections by reason. Data quality sits next to operational health instead of in a separate silo nobody opens.

Here's one representative run of 1,000 rows:

• 162 accepted, 838 rejected — a 16.2% accept rate.
• Bad phone format 348 · invalid email 193 · invalid zip 109 · empty name 106 · out-of-range age 56 · empty city 26.
• All 838 sitting in rejects/ with the rule they broke, queryable in Athena.

The low accept rate is deliberate: the Faker generator corrupts several fields independently, so a row must survive all six checks. The exact number moves run to run; the shape holds — phone and email are the strictest formats, so they dominate.

What this is worth outside the engineering team

It's worth saying plainly what the rejection lineage buys, because it isn't an engineering nicety.

An audit gets an answer instead of an investigation. When someone asks why a particular customer isn't in the quarterly report, the answer is a rule name and a timestamped object in S3 — not a two-day archaeology exercise across logs that have already rotated.

Root cause points upstream, with a number attached. 348 rows failing the phone format isn't a pipeline bug. It's a source system emitting data that violates the agreed shape, and now it's a figure you can take to the team or vendor producing it. Without the lineage, that signal is indistinguishable from "the pipeline ran fine."

Degradation shows up as a slope, not a surprise. Because every run's report lands in the quality/ zone, accept_rate_over_time is a query anyone can run. A source that quietly starts sending malformed postcodes becomes visible as a trend line — before someone notices a dashboard is wrong.

Most organisations find out about data quality problems from a business user. This inverts that.

The warehouse end: bulk, idempotent, and gated

The loader uses psycopg2.execute_values rather than row-by-row inserts, with ON CONFLICT (user_id) DO NOTHING, and reads cur.rowcount afterwards so the log distinguishes actually inserted from skipped as duplicate. CREATE DATABASE — which can't be parameterised — goes through sql.Identifier instead of string formatting. zip_code stays TEXT, because a 5-digit postcode in an INTEGER column silently loses its leading zero.

On top of that, dbt builds a silver view (stg_users, adding email_domain and age_band) feeding two marts (users_by_city, users_by_age_band). Then a fifth DAG task runs dbt test — not_null, unique, accepted_values. A bad load fails the DAG instead of quietly publishing broken marts.

A Streamlit dashboard sits over the marts, with a demo mode that synthesises the same shapes so the BI layer is reviewable without standing up the whole stack.

Two planes, deliberately separated

Infrastructure and the pipeline have different lifecycles, so they're driven by different tools:

• Control plane — Terraform. The data-lake bucket (encrypted, versioned, public access blocked, with a lifecycle rule expiring the raw zone at 30 days), a least-privilege IAM user that can only ListBucket and Get/PutObject on that one bucket — no CreateBucket, no DeleteObject — plus a Glue crawler, a CSV classifier that forces header detection, and an Athena workgroup with three saved queries.
• Data plane — Airflow. Five tasks, manual trigger only, catchup=False. The ETL never runs from CI.

A one-time bootstrap creates the remote state bucket, a DynamoDB lock table, and a GitHub OIDC deployer role. PRs touching infra/terraform/** get a read-only plan; the real apply is a manual button gated by a production environment approval. There are no long-lived AWS keys in GitHub at all — and the pipeline identity and the deployer identity are two separate principals that never share credentials.

What CI actually gates

Four jobs, plus a fifth workflow:

• lint — ruff, then python scripts/data_contract.py --check for doc drift.
• test — 56 PySpark/pytest tests on Java 17: the contract, the transform, the reject provenance, the DQ report, the loader, the generator's anomaly injection.
• smoke — spins up a real postgres:16 service container, runs the loader against a CSV fixture, and asserts the rows landed.
• dag-validate — 7 DagBag integrity tests, isolated because Airflow's pins would fight Spark's.
• gitleaks — secret scanning across the full git history, not just HEAD, plus the same check as a pre-commit hook.

63 tests total. The split isn't tidiness. The test job never installs Airflow at all, so test_dag_integrity.py guards itself with pytest.importorskip and quietly sits out that run. Resolving Airflow 2.11 against its official constraints file is fragile enough that pinning the Spark provider version alongside it makes the resolve impossible — so the DAG checks get their own job with their own dependency set. In the runtime image the two live together happily; it's CI's constraint resolution that objects.

Three things that fought back

Spark on Apple Silicon. The first image installed an x86-only JDK, and spark-submit died on arm64 with a qemu-x86_64: Could not open '/lib64/ld-linux-x86-64.so.2' loader error that takes a while to trace to the base image. The fix: an arch-aware Temurin install that picks aarch64 or x64 at build time. I also defaulted the transform to in-process local[*], with the standalone cluster left opt-in — portability beat distribution for a laptop demo.

Athena and pretty-printed JSON. I wrote the DQ report as pretty-printed, multi-line JSON. It read beautifully and broke Athena instantly with HIVE_CURSOR_ERROR — the JSON SerDe reads one object per line. Single-line NDJSON fixed it.

dbt's dependency footprint. Installing dbt-postgres on its own let the resolver pull dbt-core 2.0 — the new Fusion engine, which doesn't support the Postgres adapter (InvalidConfig dbt1005). dbt now lives in its own virtualenv inside the image, pinned to dbt-core~=1.8.0, and the DAG shells out to that binary. Keeping dbt out of Airflow's environment is the difference between a build that resolves and one that doesn't.

What I'd change

Rejection precedence hides multi-failure rows. A row with both a bad email and a bad phone is counted once, under whichever rule comes first. That makes the dashboard tidy and slightly dishonest. A better version records every rule a row violated.

The local CSV hop is a teaching choice, not a production one. The pipeline stages through a local file between S3 and Postgres so each ETL stage is separately observable. In production I'd read S3 directly into Spark and write over JDBC.

The takeaway

The single most useful decision here was refusing to let validation logic exist in more than one place. Once the contract became the source of truth, rejection lineage, PII classification, and documentation stopped being three separate chores and became things I got for free.

The pipeline doesn't just move data. It can always tell you what it refused to move, and why.

Full repo — architecture diagrams, the Terraform/OIDC setup, and screenshots from a real run: https://github.com/theofanis-tsakanikas/contract-driven-data-pipeline

If you run pipelines in production: what happens to your rejected rows? Can you point at any one of them and name the rule that dropped it? I'd genuinely like to know how other teams handle this.

---

One of a series of write-ups on the projects in my portfolio — each one a reference implementation of the trust layer that makes data and AI safe to ship.