Pipeline Design Questions

This is the round where mid-level and senior answers diverge most, and the difference is not architecture. It is whether you talk about the second run.

The question

“Orders land as CSV in S3 from an operational database, roughly hourly. Analysts need daily revenue by country in the warehouse. Design it.”

Do not draw anything yet

The first two minutes are requirements. Each answer below changes the design:

A useful phrasing: “Before I design this — is this closer to 50 MB a day or 50 terabytes? And can an order be amended after it is first written?” Two questions, and the interviewer now knows you have built one of these.

Assume the answers: ~40 GB/day, next-morning latency, orders can be amended for 72 hours, order_id is the natural key, finance reconciles the totals.

The weak answer

“I’d use Airflow to run a daily job that reads the CSVs from S3, transforms them with Spark, and writes to Snowflake. I’d add a Slack alert on failure.”

Nothing there is wrong. It scores mid-level, because every sentence is the happy path. The follow-ups it invites — and does not answer:

• What happens when it runs twice?
• What happens to the order amended 40 hours after it landed?
• The 03:00 run failed and you fixed it at 11:00. What do you run?
• The source added a column. What breaks?
• Finance says Tuesday is £4,000 short. How do you find out why?

The strong answer

Same architecture, and every one of those answered before it is asked.

S3 landing/date=YYYY-MM-DD/*.csv
        │  (1) freshness + volume gate
        ▼
bronze.orders_raw        append, typed, _ingested_at + _source_file
        │  (2) dedup on order_id, quarantine bad rows
        ▼
silver.orders            one row per order, latest version
        │  (3) join dims, aggregate
        ▼
gold.daily_revenue       one row per (day, country)
        │  (4) reconciliation gate before publish
        ▼
analysts / BI

The four numbered points are the answer. Everything else is plumbing.

Demonstrate idempotency, do not assert it

The single highest-value thing you can do in this round is show the re-run.

import duckdb
con = duckdb.connect()

con.execute("""
    create table source as select * from (values
        (1001, 1, date '2026-01-04', 'completed', 25.50),
        (1002, 2, date '2026-01-04', 'completed', 12.00),
        (1003, 1, date '2026-01-04', 'returned',  40.00)
    ) t(order_id, customer_id, ordered_at, status, amount)
""")
con.execute("create table silver_orders as select * from source where false")

def load_naive(run_date):
    con.execute("insert into silver_orders select * from source where ordered_at = ?", [run_date])

load_naive("2026-01-04")
load_naive("2026-01-04")          # the retry
print(con.execute("""
    select count(*) as rows, count(distinct order_id) as ids,
           round(sum(amount), 2) as revenue from silver_orders
""").df().to_string(index=False))

 rows  ids  revenue
    6    3     155.0

Six rows for three orders, revenue doubled, and nothing errored. Say plainly: “that is what a retry does to an append, so the load has to be idempotent.”

con.execute("truncate silver_orders")

def load_idempotent(run_date):
    con.execute("begin transaction")
    con.execute("delete from silver_orders where ordered_at = ?", [run_date])
    con.execute("insert into silver_orders select * from source where ordered_at = ?", [run_date])
    con.execute("commit")

for _ in range(3):
    load_idempotent("2026-01-04")
print(con.execute("""
    select count(*) as rows, count(distinct order_id) as ids,
           round(sum(amount), 2) as revenue from silver_orders
""").df().to_string(index=False))

 rows  ids  revenue
    3    3     77.5

Three runs, three rows. Two details to name while writing it: the transaction, so a crash between the delete and the insert rolls back rather than leaving the partition empty, and the fact that the delete predicate and the insert predicate must be identical — a mismatch is the classic source of duplicates that survive a delete-insert.

Late-arriving data

Orders can be amended for 72 hours, so partition-replacement alone loses updates:

con.execute("""
    update source set status = 'refunded', amount = 0.00 where order_id = 1002
""")
con.execute("insert into source values (1004, 3, date '2026-01-03', 'completed', 8.75)")

load_idempotent("2026-01-04")
print(con.execute("select order_id, status, amount from silver_orders order by order_id").df().to_string(index=False))

 order_id    status  amount
     1001 completed    25.5
     1002  refunded     0.0
     1003  returned    40.0

Order 1002’s amendment was picked up because it is in the 4th’s partition. Order 1004 — which arrived late and belongs to the 3rd — was not, and never will be. The fix is a lookback window plus a merge on the natural key:

con.execute("truncate silver_orders")
con.execute("insert into silver_orders select * from source where ordered_at < date '2026-01-03'")

def load_with_lookback(run_date, lookback_days=3):
    con.execute("""
        merge into silver_orders t
        using (select * from source
               where ordered_at >= ?::date - interval (?) day
                 and ordered_at <= ?::date) s
        on t.order_id = s.order_id
        when matched then update set status = s.status, amount = s.amount
        when not matched then insert by name
    """, [run_date, lookback_days, run_date])

for _ in range(2):
    load_with_lookback("2026-01-04")
print(con.execute("select order_id, ordered_at, status, amount from silver_orders order by order_id").df().to_string(index=False))

 order_id ordered_at    status  amount
     1001 2026-01-04 completed    25.5
     1002 2026-01-04  refunded     0.0
     1003 2026-01-04  returned    40.0
     1004 2026-01-03 completed    8.75

The late order is in, the amendment is applied, and running it twice changes nothing. Size the window from the worst observed arrival delay, not the typical one — and say that, because “72 hours means I’d use a 4-day window with a margin” is a senior sentence.

The backfill question

“The 03:00 run failed. You fix it at 11:00. What do you run?”

The answer is a consequence of the design above, and it is worth stating as three properties:

from datetime import date, timedelta

def backfill(start, end):
    day = start
    while day <= end:
        load_with_lookback(day.isoformat())
        print(f"  {day} reloaded")
        day += timedelta(days=1)

backfill(date(2026, 1, 3), date(2026, 1, 4))
print(con.execute("select count(*) as rows, round(sum(amount),2) as revenue from silver_orders").df().to_string(index=False))

  2026-01-03 reloaded
  2026-01-04 reloaded
 rows  revenue
    4     74.25

• Resumable — a failure on day 5 of 30 does not lose days 1-4.
• Parallel — days are independent, so they can run at once (bounded, or you take the source down).
• Retryable — re-running one day replaces that day and touches nothing else.

The contrast worth drawing: a single INSERT ... WHERE date BETWEEN has none of those properties, and it is what the weak answer implies.

The gates

An unblocked pipeline publishes whatever it computed. Name the four checks:

def run_gates(run_date):
    checks = {}
    checks["freshness"] = con.execute(
        "select max(ordered_at) >= ?::date from source", [run_date]).fetchone()[0]
    checks["volume"] = con.execute("""
        select count(*) >= 1 from source where ordered_at = ?::date
    """, [run_date]).fetchone()[0]
    checks["unique_key"] = con.execute(
        "select count(*) = count(distinct order_id) from silver_orders").fetchone()[0]
    src = con.execute("select round(coalesce(sum(amount),0),2) from source").fetchone()[0]
    tgt = con.execute("select round(coalesce(sum(amount),0),2) from silver_orders").fetchone()[0]
    checks["reconciles"] = abs(src - tgt) < 0.01

    for name, ok in checks.items():
        print(f"  {'PASS' if ok else 'FAIL'}  {name}")
    if not all(checks.values()):
        raise SystemExit(f"gate failed: {[k for k, v in checks.items() if not v]}")
    print("  gates passed — publishing")

run_gates("2026-01-04")

  PASS  freshness
  PASS  volume
  PASS  unique_key
  FAIL  reconciles
gate failed: ['reconciles']

The reconciliation gate caught a real gap — source holds an order outside the loaded window. That is the point: a gate that has never failed is not a gate, and being able to show one firing is stronger than describing it.

The four worth naming every time:

Volume is the one most candidates omit, and it catches the failure where every row that arrived is individually valid.

Monitoring, and who gets paged

“How would you know it broke?”

con.execute("""
    create table pipeline_runs (
        run_date date, started_at timestamp, finished_at timestamp,
        rows_written bigint, status varchar
    )
""")
con.execute("""
    insert into pipeline_runs values
        ('2026-01-01', '2026-01-02 03:00:00', '2026-01-02 03:04:12', 4820, 'success'),
        ('2026-01-02', '2026-01-03 03:00:00', '2026-01-03 03:03:58', 4712, 'success'),
        ('2026-01-03', '2026-01-04 03:00:00', '2026-01-04 03:04:41', 4902, 'success'),
        ('2026-01-04', '2026-01-05 03:00:00', '2026-01-05 03:00:22',    0, 'success')
""")
print(con.execute("""
    select run_date, rows_written, status,
           round(avg(rows_written) over (order by run_date rows between 3 preceding and 1 preceding)) as trailing_avg,
           datediff('second', started_at, finished_at) as seconds
    from pipeline_runs order by run_date
""").df().to_string(index=False))

   run_date  rows_written   status  trailing_avg  seconds
 2026-01-01          4820  success           NaN      252
 2026-01-02          4712  success        4820.0      238
 2026-01-03          4902  success        4766.0      281
 2026-01-04             0  success        4811.0       22

The last run succeeded with zero rows in 22 seconds. Exit code zero, no alert, and the dashboard shows a collapse the next morning. Alert on:

• a run that failed after its retries,
• a success with zero rows — the one everybody forgets,
• volume below half the trailing average,
• a runtime well above trend (a job about to breach its window),
• freshness of the output table, checked independently of the pipeline.

That last one matters: if the scheduler never fires, a pipeline-side alert never fires either. Checking max(ordered_at) in the warehouse from a separate monitor catches “the DAG was paused three weeks ago and nobody noticed”.

Schema change

“The source team adds a column and renames another. What happens?”

The answer has two halves. Additive changes should flow through — land the raw file unchanged so a new column is available when someone wants it. Breaking changes should stop the pipeline loudly:

try:
    con.execute("""
        select order_id, customer_id, ordered_at, status, amount_gbp from source limit 1
    """)
except Exception as e:
    print(type(e).__name__ + ":", str(e).split("\n")[0])

BinderException: Binder Error: Referenced column "amount_gbp" not found in FROM clause!

Prefer this to a schema-inference reader that would silently give you NULL. Then say the organisational half: a data contract checked in the producer’s CI, so a breaking change fails their build rather than your 03:00 run. That is the answer that gets remembered.

What to say about tools

Name a choice, give a reason, name the alternative:

“Airflow for orchestration because the team already runs it and the operational burden is known — though Dagster’s asset model fits this better, since the freshness and lineage story is built in rather than bolted on. Spark for the transform at 40 GB/day; below about 10 GB I would do it in the warehouse with SQL and skip the cluster entirely. dbt for the silver and gold layers, because dbt build skips downstream models when a test fails, which is the gating behaviour I described.”

Three tools, three reasons, one alternative, and a threshold. Compare with “I’d use Airflow, Spark and dbt” — same tools, no evidence of a decision.

The scoring

Practice

 rows  ids  revenue
    6    3     155.0

without lookback: 3 rows (order 1004 missing, permanently)
with lookback:    4 rows

  PASS  freshness
  FAIL  reconciles
gate failed: ['reconciles']

 2026-01-04  0  success  trailing_avg 4811.0  22s

Next: data modelling — grain, keys, and the slowly changing dimension question.