ETL Test Data: Plant Known Bugs, Assert Exact Pipeline Output

Most pipeline bugs are transform bugs. A dedup keeps the oldest version instead of the newest. An inner join drops orders whose customer is missing, and nobody notices. A lenient cast turns 2.753,71 into NULL, and revenue drops by a few thousand. Good ETL test data makes these bugs visible by planting each one on purpose, at an exact count, so you know the correct output before the pipeline runs: 506 raw customers should become 500, 14 orders should be rejected for three named reasons, and every other row should come through.

This post builds that dataset from a fixed seed and maps dbt's four built-in generic tests (unique, not_null, accepted_values, relationships) to planted violations, so you can check that each test actually fires. Then it asserts the pipeline's output against the numbers you planted. JsonFabrica generates the data and returns it as JSON over HTTP. Everything after that is your own tooling: jq converts it, your loader lands it, and dbt runs it. JsonFabrica has no connectors and doesn't run dbt.

This isn't about schema changes that break on existing rows, which test data for database migrations covers, or about generic edge values in application code, which boundary value test data generation covers. Here the target is transform logic.

Why production extracts and hand-written fixtures make poor ETL test data

The two usual sources of data pipeline test data both fail in the same way: you can't say in advance what the correct output is.

Production extracts are full of personal data, so they need masking, access controls, and a retention policy before they can go near CI. Masking isn't the same as anonymization, and that is reason enough to avoid them. They also change with every refresh. Last week's extract had 11 duplicate customers and this week's has 14, so any test that asserts an exact count breaks for reasons that have nothing to do with your code. You end up with assertions like "fewer rows than the input," which a pipeline that drops half its data also passes.

Hand-written CSV fixtures are stable, but they're small and clean. Twelve rows written by the person who also wrote the dedup logic will contain the cases that person thought of. Real sources send the cases nobody thought of: ' Active ' with padding, a timestamp with a +02:00 offset, an order whose customer was never exported.

What you want is in between: a realistic volume of mostly clean rows, with specific defects planted at exact counts and at ids you can name.

Plan the defects and their expected counts first

Write the plan before the templates. Each planted defect gets a count, fixed ids, the check that should catch it, and the expected result. The example pipeline has raw.customers and raw.orders, staging models that dedupe customers and classify orders, and an incremental fct_orders:

From the table you can already write every expected number: 1,005 raw orders, 14 rejected, 991 in fct_orders, and nothing else missing.

Generate the fixtures from a template with exact defect counts

The trick is to generate one document per table and build the rows in a loop. A for loop gives each row an index, and the index does two jobs. It becomes the row's id (padded with appendBefore, so row 7 is C00007), and it decides which rows get a defect. Template expressions have comparison and boolean operators but no arithmetic, so the bands are plain index ranges such as getVar('i') >= 491 && getVar('i') <= 493. Note that the loop variable is read with getVar('i'). A bare i in a condition is just the string "i".

The customers template:

{"rows": [<for(i, 1, 500)><if(getVar('i') > 1)>,<endIf>
<setVar('status', getRandomElement('active', 'active', 'churned'))>
<if(getVar('i') >= 491 && getVar('i') <= 493)><setVar('status', ' Active ')>
<elseIf(getVar('i') >= 494 && getVar('i') <= 496)><setVar('status', 'ACTIVE')>
<elseIf(getVar('i') >= 497)><setVar('status', 'active ')>
<endIf>
{
  "customer_id": "C<appendBefore('0', getVar('i'), 5)>",
  "email": "<toLowerCase(getRandomName())>.<getVar('i')>@example.com",
  "status": "<getVar('status')>",
  "country": "<getRandomCountry()>",
  "updated_at": "<getRandomDate('2025-01-01', '2026-07-01')>"
}<end_for>
<for(d, 1, 6)>,
{
  "customer_id": "C<appendBefore('0', getVar('d'), 5)>",
  "email": "updated.<getVar('d')>@example.com",
  "status": "churned",
  "country": "<getRandomCountry()>",
  "updated_at": "<getRandomDate('2026-07-01', '2026-09-01')>"
}<end_for>
]}

The first loop writes 500 customers. Most get a random status from getRandomElement, which picks uniformly, so listing 'active' twice makes it twice as likely. Rows 491 to 500 get one of the three planted variants instead. Because the bands are fixed, you get exactly 3 distinct bad values across exactly 10 rows, which matters for accepted_values below.

The second loop re-sends customers 1 to 6. Their updated_at ranges start where the first loop's range ends, so the re-sent row is always the newer version. Their email starts with updated., so a test can tell which version survived. The <if(getVar('i') > 1)>,<endIf> guard puts a comma before every row except the first. The second loop always writes its comma because it comes after 500 rows.

The orders template plants its defects the same way:

{"rows": [<for(i, 1, 1000)><if(getVar('i') > 1)>,<endIf>
<setVar('cust', appendBefore('0', getRandomNumber(1, 500), 5))>
<setVar('amount', formatNumber(getRandomNumber(5, 900, 2), 2))>
<setVar('placed', getRandomDate('2026-03-01', '2026-04-01'))>
<if(getVar('i') >= 11 && getVar('i') <= 14)>
<setVar('amount', formatNumber(getRandomNumber(1000, 5000, 2), 2, ',', '.'))>
<elseIf(getVar('i') >= 15 && getVar('i') <= 17)>
<setVar('placed', '2026-03-15T01:30:00+02:00')>
<endIf>
{
  "order_id": "O<appendBefore('0', getVar('i'), 6)>",
  "customer_id": <if(getVar('i') <= 3)>null
    <elseIf(getVar('i') <= 10)>"C99999"
    <else>"C<getVar('cust')>"<endIf>,
  "amount": "<getVar('amount')>",
  "placed_at": "<getVar('placed')>"
}<end_for>
]}

Each planted value is a literal or a narrow range:

• Null join keys are a literal null for rows 1 to 3.
• Orphans are the literal id C99999 for rows 4 to 10. Customer ids only go up to C00500, so it can never match.
• Clean foreign keys pick a random customer number from 1 to 500, so they always match a real customer.
• The type-coercion rows use formatNumber with a comma decimal separator and a dot for thousands, producing strings like 2.753,71. Every other amount looks like 57.92.
• The time zone rows are the literal 2026-03-15T01:30:00+02:00, which is 23:30 UTC on March 14. A model that takes the first ten characters of the string, or truncates to a day in a session time zone ahead of UTC, such as Europe/Berlin, files them under March 15. A CI runner on US time lands them on March 14 by accident and hides the bug, so pin the session to a zone east of UTC when you test.

Late arrivals get their own small template, looping over the next id range and placing orders early in the month:

{"rows": [<for(i, 1001, 1005)><if(getVar('i') > 1001)>,<endIf>
{
  "order_id": "O<appendBefore('0', getVar('i'), 6)>",
  "customer_id": "C<appendBefore('0', getRandomNumber(1, 500), 5)>",
  "amount": "<formatNumber(getRandomNumber(5, 900, 2), 2)>",
  "placed_at": "<getRandomDate('2026-03-02', '2026-03-04')>"
}<end_for>
]}

A few notes on the design. Ids come from the loop index rather than from sequences, so they're the same in every run. Sequences are durable counters that keep advancing between runs. Batch relations aren't needed here either. They always point at real generated parents, so they can't produce an orphan anyway, and loop-index ids let the orders template refer to customers by number. Each document is generated within the engine's per-generation limits: 2 seconds, 100,000 node evaluations, 10,000 loop iterations, and 8 MiB of output. The 1,000-row orders template uses about 84,000 node evaluations, so it's close to the cap. For more rows, add another alias whose template loops over the next id range.

Save the three templates with POST /v1/templates and note their ids.

One batch request, one seed, byte-identical fixtures

Request all three documents in one batch with a fixed seed:

{
  "seed": 20260930,
  "documents": [
    { "templateId": "tpl_etl_customers", "alias": "customers", "count": 1 },
    { "templateId": "tpl_etl_orders", "alias": "orders", "count": 1 },
    { "templateId": "tpl_etl_late_orders", "alias": "late_orders", "count": 1 }
  ]
}

That's three documents in total. Batches under 50 documents run synchronously and come back as 200 with the results inline, grouped by alias: the customer rows are at results.customers[0].rows. Bigger batches return 202 with a batchId you poll at GET /v1/batches/{batchId}. The response echoes the seed.

Each document's random values come from the batch seed, the alias, and the document's position, and these templates use no sequences. So the same seed returns the same bytes every time: the same random customer numbers, the same amounts, the same timestamps. If a CI run fails, you can rerun it locally against identical input. Reproducible test data from seeds covers capturing and replaying seeds in more detail. The planted counts don't depend on the seed at all. They're set by the index bands, so changing the seed changes the filler but never the expected numbers.

Land the test data in your raw layer

This step is yours, not JsonFabrica's. It doesn't write files, S3, Kafka, or warehouse tables. Convert each table's rows to newline-delimited JSON with jq:

for t in customers orders late_orders; do
  jq -c ".results.${t}[0].rows[]" batch-result.json > "$t.ndjson"
done

Then load them however your raw layer is loaded. The examples here use DuckDB with the dbt-duckdb adapter, so the whole loop runs on a laptop. Every column lands as VARCHAR, the way many raw layers store data. Typing is the pipeline's job, and that's where the coercion bug lives.

-- land-initial.sql
CREATE SCHEMA raw;

CREATE TABLE raw.customers AS
SELECT * FROM read_json('customers.ndjson', format = 'newline_delimited',
  columns = {customer_id: 'VARCHAR', email: 'VARCHAR',
             status: 'VARCHAR', country: 'VARCHAR', updated_at: 'VARCHAR'});

CREATE TABLE raw.orders AS
SELECT *, current_timestamp AS _loaded_at
FROM read_json('orders.ndjson', format = 'newline_delimited',
  columns = {order_id: 'VARCHAR', customer_id: 'VARCHAR',
             amount: 'VARCHAR', placed_at: 'VARCHAR'});

-- land-late.sql: runs after the first dbt build
INSERT INTO raw.orders
SELECT *, current_timestamp
FROM read_json('late_orders.ndjson', format = 'newline_delimited',
  columns = {order_id: 'VARCHAR', customer_id: 'VARCHAR',
             amount: 'VARCHAR', placed_at: 'VARCHAR'});

On other stacks, the same NDJSON files go to bq load --source_format=NEWLINE_DELIMITED_JSON, to a Snowflake stage and COPY INTO, or to an object storage prefix your ingestion job already watches. Use whatever your production pipeline reads from, so the test covers the real ingestion path.

The dbt models under test

To test dbt models with synthetic data, point your sources at the raw tables you just loaded. Nothing in the models needs to know the data is synthetic. The staging model for customers keeps the newest version of each customer and normalizes status:

-- models/staging/stg_customers.sql
with ranked as (
    select
        customer_id,
        email,
        lower(trim(status)) as status,
        cast(updated_at as timestamptz) as updated_at,
        row_number() over (
            partition by customer_id
            order by cast(updated_at as timestamptz) desc
        ) as version_rank
    from {{ source('raw', 'customers') }}
)

select customer_id, email, status, updated_at
from ranked
where version_rank = 1

The orders staging model uses a left join, so no order disappears, and it labels every row that can't go further:

-- models/staging/stg_orders.sql
select
    o.order_id,
    o.customer_id,
    try_cast(o.amount as decimal(12, 2)) as amount,
    cast(o.placed_at as timestamptz) as placed_at,
    o._loaded_at,
    case
        when o.customer_id is null then 'null_customer_id'
        when c.customer_id is null then 'unknown_customer'
        when try_cast(o.amount as decimal(12, 2)) is null then 'bad_amount'
    end as reject_reason
from {{ source('raw', 'orders') }} as o
left join {{ ref('stg_customers') }} as c
    on o.customer_id = c.customer_id

orders_rejected selects the rows where reject_reason is not null. fct_orders is incremental and takes the rest:

-- models/marts/fct_orders.sql
{{ config(materialized='incremental', unique_key='order_id') }}

select
    order_id,
    customer_id,
    amount,
    placed_at,
    cast(placed_at at time zone 'UTC' as date) as order_date_utc,
    _loaded_at
from {{ ref('stg_orders') }}
where reject_reason is null
{% if is_incremental() %}
  -- not placed_at > max(placed_at): late orders have old timestamps
  and _loaded_at > (select max(_loaded_at) from {{ this }})
{% endif %}

Cast syntax and time zone functions differ between warehouses (BigQuery uses SAFE_CAST, for example), so adapt the SQL. The test data and the expected numbers stay the same. Whether your warehouse rejects 2.753,71 or reads it as 2.753, the reject count and the revenue total below will catch it.

Proving dbt's generic tests actually fail

A test that has never failed hasn't been tested. It might point at the wrong column, have a typo in its accepted values, or be disabled by a config you forgot about. Planted data lets you check each one. Declare the four generic tests on the raw sources, give each a name, and set them to warn so they report without stopping the build:

# models/sources.yml
version: 2

sources:
  - name: raw
    schema: raw
    tables:
      - name: customers
        columns:
          - name: customer_id
            data_tests:
              - unique:
                  name: planted_dup_customer_ids
                  config: {severity: warn}
          - name: status
            data_tests:
              - accepted_values:
                  name: planted_bad_status
                  values: ['active', 'churned']
                  config: {severity: warn}
      - name: orders
        columns:
          - name: customer_id
            data_tests:
              - not_null:
                  name: planted_null_customer_ids
                  config: {severity: warn}
              - relationships:
                  name: planted_orphan_orders
                  to: source('raw', 'customers')
                  field: customer_id
                  config: {severity: warn}

dbt versions differ in the details. The data_tests: key arrived in dbt 1.8, and older projects use tests:. dbt 1.10.5 and later also accept an arguments: block for inputs like values, to, and field, and may warn you to use it. Check the docs for the version you run.

Each test's failure count is the number of rows its query returns, and that isn't always the number of bad rows. In dbt's default implementations:

Two of those catch people out. accepted_values would report 3 even if 100 rows had bad statuses. And relationships ignores nulls, so the 3 null customer_id rows only show up in not_null. If you test join keys with relationships alone, null keys get through unreported.

In recent dbt versions, each test result in target/run_results.json carries a failures count. Compare it with a checked-in expectations file:

planted_bad_status 3
planted_dup_customer_ids 6
planted_null_customer_ids 3
planted_orphan_orders 7

Then declare the same tests on stg_customers and fct_orders at the default error severity. There they must pass. Together, the two halves show that each test catches its defect and that the transform removed exactly those rows.

Asserting the pipeline output: dedup, rejects, time zones, late data

The expected numbers are constants, taken straight from the plan table. They're not computed by rerunning the pipeline's logic:

# tests/pipeline/test_outputs.py
import json
from decimal import Decimal

import duckdb
import pytest

# Known before the run: these come from the template's index bands.
RAW_CUSTOMERS, CUSTOMERS, RAW_ORDERS = 506, 500, 1005
REJECTS = {"null_customer_id": 3, "unknown_customer": 7, "bad_amount": 4}
REJECTED_IDS = {f"O{n:06d}" for n in range(1, 15)}
OFFSET_IDS = ["O000015", "O000016", "O000017"]
LATE_IDS = [f"O{n:06d}" for n in range(1001, 1006)]


@pytest.fixture(scope="session")
def db():
    conn = duckdb.connect("etl_test.duckdb", read_only=True)
    yield conn
    conn.close()


def scalar(db, sql, params=None):
    return db.execute(sql, params or []).fetchone()[0]


def test_dedup_keeps_newest_version(db):
    assert scalar(db, "select count(*) from raw.customers") == RAW_CUSTOMERS
    assert scalar(db, "select count(*) from stg_customers") == CUSTOMERS
    assert scalar(db, """
        select count(*) from stg_customers
        where customer_id between 'C00001' and 'C00006'
          and email like 'updated.%'""") == 6


def test_status_variants_normalized(db):
    rows = db.execute("""
        select distinct status from stg_customers
        where customer_id between 'C00491' and 'C00500'""").fetchall()
    assert rows == [("active",)]


def test_every_order_kept_or_rejected(db):
    assert scalar(db, "select count(*) from raw.orders") == RAW_ORDERS
    reasons = dict(db.execute("""
        select reject_reason, count(*) from orders_rejected
        group by 1""").fetchall())
    assert reasons == REJECTS
    kept = scalar(db, "select count(*) from fct_orders")
    assert kept + sum(reasons.values()) == RAW_ORDERS


def test_offset_timestamps_bucket_to_utc_day(db):
    rows = db.execute("""
        select distinct order_date_utc::varchar from fct_orders
        where order_id in (?, ?, ?)""", OFFSET_IDS).fetchall()
    assert rows == [("2026-03-14",)]


def test_late_arrivals_reach_fact_table(db):
    assert scalar(db, """
        select count(*) from fct_orders
        where order_id in (?, ?, ?, ?, ?)""", LATE_IDS) == 5


def test_revenue_matches_fixture(db):
    with open("batch-result.json") as f:
        results = json.load(f)["results"]
    orders = results["orders"][0]["rows"] + results["late_orders"][0]["rows"]
    expected = sum(Decimal(o["amount"]) for o in orders
                   if o["order_id"] not in REJECTED_IDS)
    assert scalar(db, "select sum(amount) from fct_orders") == expected

test_every_order_kept_or_rejected is the reconciliation check, and it matters most. An inner join also produces 991 rows in fct_orders, so the row count alone can't tell a correct pipeline from one that drops data. The check that kept rows plus rejected rows equals raw rows, with the reject reasons matching the plan, can. The revenue test reads the same JSON that was loaded and sums every order except the 14 planted rejects. A lenient cast that turned a valid amount into NULL, or read 2.753,71 as 2.753, would change that sum.

The tests assume dbt builds into DuckDB's default main schema, so model names resolve without a prefix.

Running the ETL test in CI

The order of the steps matters because of the late arrivals. The first build has to run before they land, or the incremental filter never gets tested:

#!/usr/bin/env bash
# etl-test.sh: generate, land, build, land late rows, build, assert
set -euo pipefail
API=https://api.jsonfabrica.com/v1
rm -f etl_test.duckdb  # dbt profile: dbt-duckdb, path etl_test.duckdb

# 1. Generate (three documents, so it returns 200 with results).
curl -fsS -X POST "$API/batches" \
  -H "Authorization: Bearer $JSONFABRICA_API_KEY" \
  -H 'content-type: application/json' \
  -d @etl-fixtures/batch.json > batch-result.json
jq -e '.status == "completed"' batch-result.json > /dev/null \
  || { cat batch-result.json >&2; exit 1; }
echo "seed=$(jq .seed batch-result.json)"

# 2. Land with your own tooling, then run the first build.
for t in customers orders late_orders; do
  jq -c ".results.${t}[0].rows[]" batch-result.json > "$t.ndjson"
done
duckdb etl_test.duckdb < etl-fixtures/land-initial.sql
dbt build

# 3. Late arrivals land after the first run; the incremental run follows.
duckdb etl_test.duckdb < etl-fixtures/land-late.sql
dbt build

# 4. Did each planted test fail by exactly the planted amount?
jq -r '.results[] | select(.unique_id | contains(".planted_"))
  | "\(.unique_id | split(".")[2]) \(.failures)"' \
  target/run_results.json | sort > planted-actual.txt
diff etl-fixtures/planted-expected.txt planted-actual.txt

# 5. Did the transforms produce the planned output?
pytest tests/pipeline

dbt build exits non-zero if an error-severity test on the staging models fails. The warn-level source tests don't stop the build, and the diff checks them instead. It fails if a planted test didn't run at all, because its line is missing, or if it caught a different number of rows than planned.

When a production incident turns up a new shape, such as a status with a trailing tab or an order with a negative amount, add a band to the template and a row to the plan table. After that, the next change to the transform is tested against that shape too.

FAQ

How do you test an ETL pipeline? Load an input dataset whose properties you know exactly, such as how many duplicate keys, orphaned foreign keys, null join keys, and malformed values it contains, then run the pipeline and assert the output against numbers you wrote down before the run: rows after deduplication, rows rejected and why, and totals. Also check that every input row is either in the output or in a reject table, so nothing is dropped silently. Run it from a fixed seed in CI on every change to the transforms.

How do I generate test data for dbt models? Generate the raw tables your sources point at, not the models themselves. Use a generator that lets you plant specific defects at exact counts and fixed ids, land the rows in your raw schema with your usual loader, then run dbt build against them. Because you planted the defects, you know what each staging model and test should produce before you run it.

How do I know my dbt tests actually work? Give each test at least one input it must fail on. Declare the same generic tests on your raw sources with warn severity, plant a known number of violations in the raw data, and check that each test's failure count in run_results.json matches the planted number. The same tests on your staging models should then pass, which shows the transform fixed exactly those rows.

Should I use production data to test data pipelines? Usually not as your main fixture. Production extracts contain personal data, change every time you refresh them, and don't tell you how many bad rows they contain, so you can't write exact expected counts against them. Synthetic data with planted defects gives stable, known answers. Keep production-derived checks, such as data tests on the real warehouse, as monitoring rather than as your pipeline's test suite.

Can JsonFabrica load test data into Snowflake, BigQuery, or S3? No. JsonFabrica returns generated JSON over HTTP and has no connectors. It doesn't write files, object storage, message queues, or warehouse tables, and it doesn't run dbt, Airflow, or Spark. You convert the JSON with a tool such as jq and load it with your own loader, for example DuckDB, bq load, Snowflake COPY INTO, or an S3 upload.

What is the difference between dbt unit tests and testing with synthetic data? dbt unit tests, available in dbt 1.8 and later, check one model's SQL against a few mocked input rows defined in YAML, before the model is built. Synthetic raw data tests the whole DAG at realistic volume, including joins across models, incremental runs, and your data tests. Unit tests are good for pinning one tricky expression. A planted dataset shows the pipeline as a whole produces the right counts.

A pipeline test is only as good as what you know about its input. JsonFabrica's batch API generates ETL test data with duplicates, orphans, null keys, and malformed values planted at exact counts and fixed ids, reproducible from one seed. Your own tooling lands it and runs the pipeline.