Disappearing Ink

An easy Pipeline Design interview practice problem on DataDriven. Design the pipeline on the canvas: sources, transforms, storage and checks, then get feedback on it.

Domain

Pipeline Design

Difficulty

easy

Seniority

mid

Asked in

Snap

Problem

An analytics lead keeps fielding the same complaint: a dozen teams need data shapes that the ingest already threw away, while about 10 billion engagement events a day (opens, replays, story views, ad views) keep arriving. Design the pipeline that builds the curated tables those teams query every morning.

Business requirements

• Our teams keep changing what they want from the data, so we do not want to decide the final shape before we even store it.
• A dozen teams query this data, and none of them should ever have to read the raw feed directly.
• When a day's data is malformed I do not want the analysts to be the ones who discover it.
• This has to run every day on its own and page someone when it does not.

Worked solution and explanation

What this really is

This is land-first ELT dressed up as a 10-billion-events-a-day scale problem. A dozen teams want different, shifting shapes, so no transform can be locked in before storage. Anyone can draw raw storage, Spark and a warehouse. The trap is publishing the transform output straight into the table teams read and treating the quality check as an alert. One bad daily partition then overwrites the live table, and the check only tells you after a dozen teams have queried it.

The other trap is transforming at ingest. Curate on the way in and the raw detail is gone, so the first team that wants a dropped field means re-collecting data you cannot get back. Land raw, and that request becomes a reprocessing job.

Walk the requirements

Step 1: Land raw events untouched

Write the stream to cheap object storage, partitioned by date, with no shaping and no dropped fields. Storage is cheap and re-collection is impossible, so this is the safety net behind every later shape.

Step 2: Rebuild curated tables in a daily batch

A Spark job reads yesterday's raw partition only. The analysts work off yesterday's data, so a streaming design would buy freshness nobody asked for. Mapping, sessionizing and dropDuplicates live here, downstream of storage, where changing them is cheap and re-runs are idempotent.

Step 3: Gate between staging and publish

The transform writes to a staging path. The gate checks for an empty partition, null user_id and a broken schema, and only then does publish touch the curated table. Because the gate is wired into the task graph, a failure halts the publish instead of merely reporting it.

Step 4: Orchestrate the chain and page on failure

The scheduler owns the daily run, retries transient failures and pages on-call when a run is late or dead. A cron job with no owner satisfies none of that.

# Airflow DAG: land-first, transform-late, gate-before-publish.
# The order of the >> chain IS the architecture: raw storage is
# upstream of the transform, and the quality gate sits between the
# transform and the curated publish so bad partitions never reach teams.
from airflow import DAG
from airflow.operators.python import PythonOperator, ShortCircuitOperator
from datetime import datetime

def transform_day(ds, **_):
    # Read ONLY yesterday's raw partition, not the whole lake, so
    # compute stays proportional to one day (~8 TB) not all history.
    raw = spark.read.json(f"s3://events-raw/dt={ds}/")
    curated = (
        raw
        .transform(map_known_fields)     # tolerate unknown fields on drift
        .transform(sessionize)
        .dropDuplicates(["event_id"])    # idempotent re-runs
    )
    # Stage to a side path; publish only after the gate passes.
    curated.write.format("delta").mode("overwrite").save(f"s3://curated-staging/dt={ds}/")

def quality_gate(ds, **_):
    staged = spark.read.format("delta").load(f"s3://curated-staging/dt={ds}/")
    checks = [
        staged.count() > 0,                                  # partition not empty
        staged.filter("user_id IS NULL").count() == 0,       # required field present
        set(EXPECTED_COLS).issubset(staged.columns),         # schema not broken
    ]
    return all(checks)  # ShortCircuit: False halts the DAG and alerts

def publish(ds, **_):
    spark.read.format("delta").load(f"s3://curated-staging/dt={ds}/") \
        .write.format("delta").mode("overwrite") \
        .partitionBy("dt").save("s3://curated/engagement/")

with DAG(
    "engagement_daily_elt",
    schedule="0 3 * * *",          # T+1 morning delivery
    start_date=datetime(2026, 1, 1),
    catchup=True,                  # backfill reprocesses raw into new shapes
    default_args={"retries": 2, "on_failure_callback": page_oncall},
) as dag:
    t = PythonOperator(task_id="transform", python_callable=transform_day)
    g = ShortCircuitOperator(task_id="quality_gate", python_callable=quality_gate)
    p = PythonOperator(task_id="publish", python_callable=publish)
    t >> g >> p

Publish straight from the transform

The job overwrites the curated table directly and a check runs afterward. An empty or null-ridden day is already live when the alert fires, and all twelve teams have read it.

Stage, gate, then publish

The job writes to a staging path and ShortCircuitOperator blocks publish on failure. The curated table keeps yesterday's good data until a fixed run replaces it.

Common follow-up questions

• A team needs a field you never mapped into the curated tables, but it exists in the raw events. What do you do? (Tests the ELT payoff: extend the transform and backfill from the raw landing zone, with no change to ingestion.)
• Event schemas drift weekly. How does the pipeline survive a new or removed field? (Tests drift handling: raw keeps everything, the transform tolerates unknown fields, and the gate's schema check blocks breaking changes.)