Designing data pipelines: the senior round

A repeatable answer shape

Senior design rounds reward a visible structure more than a clever tool choice. Walk the same seven steps every time, and say the trade-off out loud at each one.

1. Requirements and SLAs. Ask for freshness (how stale the newest queryable row may be, and at what percentile), volume (bytes or rows per day, peak versus average, growth), correctness (exact source state, at-least-once with deduplication, or approximate), retention, and a cost ceiling. Write the numbers down; every later choice should cite one.
2. Data contract. Who owns the schema, which fields form the key, which field orders changes, how deletes are represented, and what a producer may change without notice. A contract turns schema drift from a surprise into a policy.
3. Write path. Every write idempotent (running it twice gives the same table), replayable from an immutable raw copy, and order-aware when the source sends updates. Most correctness bugs live here.
4. Read path. Who queries what with which filters: partition and file layout, compaction, and whether consumers read a curated table or a raw one.
5. Failure and recovery. What happens when a job dies halfway, runs twice, or loads bad data for days — and how you get back.
6. Observability. Rows in and out, rows rejected, freshness lag, runtime and cost per run, each with an alert.
7. Trade-offs. Name what you gave up: latency for cost, strictness for availability, simplicity for flexibility.

Access control belongs in the design too, but be precise about the layer. Spark itself provides RPC authentication and encryption, local-storage encryption, and Web UI view and modify ACLs — and security features like authentication are not enabled by default. Spark 3.5.7 does not provide table-, row-, or column-level permissions on data. Those come from a catalog or governance layer outside Spark — for example Unity Catalog (Databricks), Apache Ranger, or AWS Lake Formation — each with its own model. In an answer, say which layer enforces each grant, and that a grant enforced only at the SQL layer does not stop a job that holds direct storage credentials from reading the files.

Check the sources: Spark security

Writes you can safely run twice

Retries, reruns, and backfills all happen in production, so every write path needs an answer to "what if this runs twice?" Four mechanisms exist in Spark and Delta Lake, and each has a trap.

• Dynamic partition overwrite. spark.sql.sources.partitionOverwriteMode defaults to static, where an overwrite of a partitioned data source table deletes every partition matching the partition spec first. In dynamic mode Spark overwrites only the partitions that receive data at runtime; the writer option partitionOverwriteMode takes precedence over the session setting, and Hive serde tables are always overwritten dynamically. On a local run, overwriting one day's rows into a three-day Parquet directory left only that day in static mode and kept all three in dynamic mode. Delta Lake 2.0 and above supports the same mode. Trap: one stray row in the wrong partition replaces that entire partition.
• Delta replaceWhere. Overwrites exactly the rows matching a predicate, atomically, and validates that every written row matches it. Locally, one row outside the predicate failed the write with DELTA_REPLACE_WHERE_MISMATCH and left the table unchanged. The Delta docs recommend it over dynamic overwrite for that reason, and the two writer options cannot be combined.
• Delta idempotent writes. Pass txnAppId and txnVersion. Delta records the pair as a transaction identifier in the log and ignores a later write from the same app ID whose version is not higher than the last recorded one. Locally, re-sending version 7 added no rows and created no commit; version 8 appended. Traps: a retry with the same version but different data is skipped silently too, and a date-derived version skipped a backfill of an older date (20260901 after 20260903). In foreachBatch, the Delta docs pair a stable app ID with the batch ID, and require a new app ID if you replace the streaming checkpoint.
• MERGE from a deterministic source. An upsert converges on replay only if the source is reduced to one row per key by a deterministic rule and the update is guarded by an ordering column. Delta refuses ambiguity: a MERGE in which several source rows matched one target row failed with DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE.

day = "2026-09-02"
rerun = spark.createDataFrame([(day, 30), (day, 31)], "day string, v int")

# Replace one day atomically; the write fails if any row falls outside the predicate.
(rerun.write.format("delta")
    .mode("overwrite")
    .option("replaceWhere", f"day = '{day}'")
    .save("orders_by_day"))

# Safe to retry: the same input produces the same table.
spark.read.format("delta").load("orders_by_day").groupBy("day").count().orderBy("day").show()

The block assumes a Delta table orders_by_day partitioned by day holding three days of rows; running it twice printed the same counts (2, 2, 1). For how these writes pair with a streaming checkpoint, see Structured Streaming; for the commit log that makes them atomic, see Delta Lake.

Check the sources: Spark: partitionOverwriteMode · Delta: writes, time travel, schema · Delta: idempotent foreachBatch writes · Delta protocol: transaction identifiers · Delta: MERGE and CDC

CDC: ordering, deletes, and schema drift

A change feed is a stream of inserts, updates, and deletes for keyed records. Spark does not read a database's change log itself: capture is a connector's job outside Spark, which typically lands events in Kafka or files. The Spark design starts where those events land, and it has to answer three questions.

• Order by a source sequence, never by arrival. Each event needs an ordering field produced by the source — a log position or a per-key version. Arrival order, file order, and ingestion time are not ordering; retries and parallel readers reorder them.
• Apply the latest change per key. Within a batch, keep one row per key with row_number() over the key ordered by sequence descending — not dropDuplicates, which keeps an arbitrary row. Across batches, guard the update with s.seq > t.seq so a late, older event cannot overwrite newer state. The Delta docs' CDC example takes the latest change per key the same way before merging.
• Keep deletes as tombstones. A hard delete forgets the key's last sequence, so a late update re-inserts it. A soft-delete row (is_deleted = true plus its sequence) lets the guard reject the stale event. Publish a current view that filters tombstones, and purge old tombstones only after the maximum lateness you accept.

The lesson example below shows both halves. In a comparison run, arrival-order dedup with hard deletes applied a late batch as-is: key 1 regressed from shipped (seq 30) to paid-late (seq 22), and deleted key 2 came back. The guarded merge with tombstones kept seq 30 and the tombstone through the late batch and a full replay of the first batch.

Schema drift. Delta rejected an append carrying a column the table lacked until the write set mergeSchema; the session setting spark.databricks.delta.schema.autoMerge.enabled does the same for all writes and is required for schema evolution in MERGE. Evolution adds columns; it did not change id from int to string (DELTA_FAILED_TO_MERGE_FIELDS). The risk of leaving auto-merge on is that a producer's typo becomes a permanent column in every downstream table. A safer contract: keep raw payloads in Bronze, alert on any new field, and promote it to Silver deliberately.

Check the sources: Delta: MERGE and CDC · Delta: writes, time travel, schema

Data quality that fails loudly without blocking

"Fail loudly but do not block" sounds contradictory until you separate rows that break a rule from batches that break a contract.

• Constraints for true invariants. Delta enforces NOT NULL columns and CHECK constraints on every write. Locally, one negative amount in a three-row append failed with DeltaInvariantViolationException: [DELTA_VIOLATE_CONSTRAINT_WITH_VALUES] CHECK constraint amount_non_negative (amount >= 0) violated by row with values: - amount : -1.0, and the two valid rows were not committed either. ALTER TABLE ADD CONSTRAINT checks existing rows first; it refused a constraint that one stored row violated. Constraints are loud and blocking, so reserve them for rules that must never be false.
• Quarantine for row-level rules. Split the batch with the rule's predicate: valid rows go to the table, failing rows go to a quarantine table with the rule name and batch ID. Consumers still get fresh data; the quarantine rate is the loud part.
• Write-audit-publish for batch-level checks. Row count against a trailing baseline, duplicate keys, null rates, referential checks: write the batch to a staging table, run the audits, and only then publish to the consumer table in one commit (an append, a replaceWhere overwrite, or a MERGE from staging). A failed audit pages someone while consumers keep reading the last good version. This is a pattern built from ordinary writes, not a single Spark API.
• Metrics and alerts. Emit rows read, written, quarantined, and rejected per batch, plus freshness lag. Alert on ratios and baselines, not only on job failure: a job that succeeds while writing zero rows is the incident nobody pages for.

Check the sources: Delta: constraints · Delta: writes, time travel, schema

Recovery: time travel, RESTORE, and replay from raw

Recovery depends on what you still have. A lakehouse offers two independent safety nets, and a senior answer uses both.

• Time travel and RESTORE. Every Delta commit is a version. versionAsOf reads an old snapshot, and RESTORE TABLE … TO VERSION AS OF n rolls back by writing a new commit, so the rollback is itself in history. Locally, after a bad overwrite replaced three rows with one, RESTORE to version 1 brought back ids 1–3, and history read WRITE, WRITE, WRITE, RESTORE.
• VACUUM bounds it. Time travel needs both the log and the data files. VACUUM deletes unreferenced data files older than the retention threshold (7 days by default, set by delta.deletedFileRetentionDuration); log history defaults to 30 days (delta.logRetentionDuration). After a zero-hour VACUUM — possible only with the retention safety check disabled — RESTORE failed with "Not all files from version 1 are available in file system", and collect() on the old version raised SparkFileNotFoundException. Yet count() on that version still returned 2: Delta answered it from log statistics without opening files, so a count does not prove a version is readable.
• Replay from immutable raw. If the source sent bad data, rolling a table back does not help; you need the original input. Keep an append-only raw landing zone retained longer than your detection window, and rebuild affected partitions downstream with idempotent writes such as replaceWhere per day.

A recovery answer runs contain, scope, repair, prove, prevent: stop the bad writer; use history to find the first bad version and every downstream table that read it; restore or rewrite the affected range; recompute downstream partitions; reconcile counts against the source; add the check that would have caught it. If VACUUM already removed the good files, only replay remains — which is why retention should exceed the time a problem can go unnoticed.

Check the sources: Delta: VACUUM and RESTORE · Delta: writes, time travel, schema

Telling an incident story with numbers

The prep advice in the source bank is to have two or three incidents ready in STAR form — a skew fix with before and after numbers, a cost or runtime reduction, and a correctness bug you caught. The structure is easy; the numbers and the depth behind them are what interviewers probe. Use your own real incidents: this page cannot supply them, and an invented story falls apart at the second follow-up.

• Situation — one or two sentences: the pipeline, its SLA, its scale (input per run, rows, cluster size).
• Task — what was broken or at risk, as a number: runtime against SLA, cost per run or month, mismatch rate, consumers affected.
• Action — diagnosis before fix. Name the evidence (the slow stage, max versus median task time, shuffle read, the plan change), then the change, then the alternatives you rejected and why.
• Result — before and after, measured the same way on comparable input: runtime, cost, data volume, error rate — plus the guard that stops a recurrence.

Per story type, gather:

• Skew fix: max and median task duration and shuffle read for the slow stage, the hot key's share of rows, whether AQE skew-join handling applied (in Spark 3.5.7 it splits skewed partitions of sort-merge and shuffled hash joins; it does not fix a skewed aggregation), and the fix you chose. See Joins and skew and the Spark UI.
• Cost or runtime reduction: one defensible unit (cost per run, node-hours, bytes scanned), where the time went before, the change with the largest effect, what it traded away, and how you ruled out smaller input as the cause.
• Correctness bug: how it surfaced, the root cause in one sentence, rows or days affected, how you repaired history and told consumers, and the automated check you added.

Check the sources: Web UI · SQL tuning: AQE

Try it on a small dataset

Run this in a PySpark notebook with an existing spark session. Create a local session if needed. The example was run for review; this website does not execute Spark.

from delta.tables import DeltaTable
from pyspark.sql import Window, functions as F

path = "customers_delta"
spark.createDataFrame(
    [(1, 10, "new", False), (2, 10, "new", False)],
    "id int, seq long, status string, is_deleted boolean",
).write.format("delta").mode("overwrite").save(path)

# One change batch: out-of-order events for id 1, a delete for id 2, an insert for id 3.
changes = spark.createDataFrame(
    [(1, 30, "u", "shipped"), (1, 20, "u", "paid"), (2, 25, "d", None), (3, 15, "c", "new")],
    "id int, seq long, op string, status string",
)


def apply_changes(batch):
    latest = (
        batch.withColumn("rn", F.row_number().over(
            Window.partitionBy("id").orderBy(F.col("seq").desc())))
        .filter("rn = 1").drop("rn")
    )
    (DeltaTable.forPath(spark, path).alias("t")
        .merge(latest.alias("s"), "t.id = s.id")
        .whenMatchedUpdate(
            condition="s.seq > t.seq",
            set={"seq": "s.seq",
                 "status": "CASE WHEN s.op = 'd' THEN t.status ELSE s.status END",
                 "is_deleted": "s.op = 'd'"})
        .whenNotMatchedInsert(values={
            "id": "s.id", "seq": "s.seq", "status": "s.status",
            "is_deleted": "s.op = 'd'"})
        .execute())


apply_changes(changes)
# A late batch carrying older events, then a replay of the first batch.
apply_changes(spark.createDataFrame(
    [(1, 22, "u", "paid-late"), (2, 12, "u", "resurrected")],
    "id int, seq long, op string, status string"))
apply_changes(changes)

spark.read.format("delta").load(path).orderBy("id").show()

+---+---+-------+----------+
| id|seq| status|is_deleted|
+---+---+-------+----------+
|  1| 30|shipped|     false|
|  2| 25|    new|      true|
|  3| 15|    new|     false|
+---+---+-------+----------+

Ran on local PySpark 3.5.7 (local[2], JDK 17, Delta Lake 3.2.1) on 2026-09-15; output above is observed.

A daily job writes to a Delta table and is sometimes retried after a partial failure. How do you make the write safe to run twice, and what can still go wrong?

Say your answer out loud before opening the notes.