Lyft Data Engineer Interview Preparation Guide - Junior Level

Direct answer

Do not start by writing a large test suite. Start by making the pipeline reproducible, because an untested pipeline that also is not reproducible means any test you write next cannot be trusted either, a failure could be the bug or could just be run-to-run noise. The order is: pin reproducibility first, add a thin layer of the highest-value tests, wire those into continuous integration (CI, the system that automatically runs checks on every code change) as a hard merge gate, add data-validation gates at the pipeline's entry points, and only then trust the result enough to deploy, starting from a shadow deployment rather than jumping straight back to full production traffic. The first sprint buys just enough of each of those to stop active bleeding; the rest of the quarter builds out the depth.

Structured elaboration

The first tests to add, in priority order.

1. A reproducibility test. Train (or run inference on) a fixed, pinned data snapshot twice with the same explicit seed and assert the two runs produce identical or near-identical output. This has to come before every other test, because a test suite layered on top of a non-reproducible pipeline produces flaky results that look exactly like real bugs, and nobody can tell the difference without this check existing first.
2. An end-to-end smoke test. Run the full training pipeline against a tiny fixture dataset (a few dozen rows, not the real data) and assert it completes without crashing, within a bounded resource budget. This catches the most common inherited-codebase failure mode, that the pipeline simply cannot run cleanly end to end anymore, before investing in anything more detailed.
3. A golden-file regression test. Freeze the current model's predictions on a fixed validation set as a recorded baseline, and assert that any future change stays within a small tolerance of that baseline unless the change is an intentional retrain. This is what actually catches an accidental behavior change introduced by a refactor, which is the single most common way a legacy codebase silently breaks itself further.
4. Unit tests on the highest-risk transformations. Given a known input, assert a known expected output, targeted specifically at label computation, join-key logic, and any feature with an obvious opportunity for leakage (a feature that could see information from after the label's outcome time). These are cheap, fast, and catch a large fraction of real defects for the effort involved.

CI changes. Every one of the tests above becomes a required, blocking check on every pull request, not an optional or manually-run script. Alongside that: pin every dependency version in a lockfile and pin the exact container or environment image the pipeline runs in, so "it works on my machine" stops being able to explain a discrepancy. Add a separate, scheduled (for example nightly) full-pipeline run against a larger held-out dataset than the fast per-change checks use, since some defects only show up at realistic data scale or after enough wall-clock drift in upstream systems. Every CI training run should automatically log its code commit, data version, and resulting metrics to a shared experiment-tracking store, so there is a trustworthy record of what actually ran, rather than relying on someone's memory or a spreadsheet.

Data-validation gates. Schema validation at ingestion (types, required fields, allowed value ranges) rejects a malformed batch before it ever reaches training. A distribution check compares each incoming batch's key feature distributions against a stored reference and flags anything that has drifted past a threshold, catching an upstream pipeline change before it silently degrades the model. A primary-key uniqueness and duplicate-row check on the training set catches the common bug where a join fans out and silently duplicates rows, inflating the apparent size and skewing the trained model toward the duplicated segment. A leakage-audit checklist, applied to every new or changed feature, requires writing down the exact timestamp each feature becomes known and confirming it is strictly before the prediction time, run as a required gate rather than an optional code-review suggestion.

Safe deployment strategy. Because none of the above exists yet at the start, do not go back to serving full production traffic once the code technically runs again. Start with a shadow deployment: the "repaired" pipeline scores real production traffic in parallel with whatever is currently serving, its predictions are logged and compared, but never returned to a user. Once the shadow comparison shows agreement with the current production behavior on the metrics that matter (or, if the goal is to fix a known regression, shows the expected improvement without new failure modes), move to a small-percentage canary with an explicit, automated rollback trigger tied to the same golden-file and data-validation checks used in CI, then ramp progressively. Skipping straight to canary or full traffic on a codebase with zero prior trust in its own outputs defeats the entire point of doing this work.

Sprint versus quarter.

Worked example

The reproducibility test from priority item 1, run against the actual inherited pattern versus the fix, on a pinned four-value input array [1.0, 2.0, 3.0, 4.0]:

import numpy as np

# The global seed is pinned here ONLY so this demonstration reproduces for a
# reader. It does not fix the bug: legacy_train still draws from the shared
# global state, so two consecutive calls advance that state and disagree.
np.random.seed(1)

def legacy_train(data):
    # BUG: draws from whatever the global numpy RNG state happens to be, no local seed
    weights_init = np.random.randn(4)
    return float((weights_init * data).sum())

def fixed_train(data, seed):
    rng = np.random.default_rng(seed)
    weights_init = rng.standard_normal(4)
    return float((weights_init * data).sum())

def reproducibility_test(train_fn, **kwargs):
    data = np.array([1.0, 2.0, 3.0, 4.0])
    a = train_fn(data, **kwargs)
    b = train_fn(data, **kwargs)
    print(f"  run A = {a:.6f}, run B = {b:.6f}")
    return bool(np.isclose(a, b, atol=1e-12))

print("legacy_train (unseeded, shared global RNG):")
print("  reproducibility test passes:", reproducibility_test(legacy_train))
print("fixed_train (local generator, seed=84):")
print("  reproducibility test passes:", reproducibility_test(fixed_train, seed=84))

Output:

legacy_train (unseeded, shared global RNG):
  run A = -5.475557, run B = -1.548062
  reproducibility test passes: False
fixed_train (local generator, seed=84):
  run A = -5.201353, run B = -5.201353
  reproducibility test passes: True

Against legacy_train the two runs are -5.475557 and -1.548062, not identical, so the test correctly FAILS against the actual defect. The identical test against fixed_train(data, seed=84) gives -5.201353 twice, so it correctly PASSES once the fix (a locally-seeded generator, no dependence on global state) is applied. Note what pinning the global seed at the top does and does not buy: it makes this demonstration reproducible across processes, and legacy_train STILL fails, because two consecutive calls advance the shared global state and the second one therefore sees different numbers than the first. A global seed is not a substitute for a local generator. That the same test fails on the real bug and passes on the real fix is what makes it trustworthy as the first gate; a reproducibility test that would pass either way would give false confidence.

Trade-offs and pitfalls

• Writing broad test coverage before reproducibility is a common wrong turn. A large test suite bolted onto a non-reproducible pipeline just produces a large number of flaky tests, which trains the team to ignore CI failures, the opposite of the goal.
• The golden-file tolerance has to be chosen deliberately. Too tight, and every legitimate retrain trips the check and gets ignored; too loose, and it stops catching real regressions. Tie the tolerance to the same significance reasoning used for any other model-comparison decision, not an arbitrary round number.
• Schema and drift gates need an escape hatch. A hard block on every ingestion anomaly will eventually block a legitimate business change (a genuinely new customer segment, a new product line); pair automatic gating with a fast, logged manual override path rather than a check nobody can bypass under real time pressure.
• Shadow deployment takes real calendar time before any canary traffic can start, and stakeholders under pressure to "just ship the fix" will push to skip it; the cost of skipping it is exactly the risk this whole plan exists to remove, so it is worth naming explicitly rather than quietly dropping under pressure.
• Trying to do all of this in the first sprint instead of triaging is the other common wrong turn. A senior answer explicitly sequences the work (reproducibility and the cheapest high-value tests first, breadth and automation across the quarter) rather than treating the whole list as one undifferentiated backlog.

Direct answer

Use named aggregation inside .groupby(...).agg(...) for readable, self-documenting column names, .nunique() for the distinct-user count, and .sort_values().head(5) for the top 5. as_index=False returns the group key as a regular column immediately, which is usually what you want for a result you are about to sort, display, or write out.

Approach

1. Group by event_type and use named aggregation to compute total_value (sum) and avg_value (mean) in one call, with clear output column names chosen up front rather than pandas' default multi-level column names.
2. A second grouped aggregation (or the same one, extended) for unique_users via .nunique() on user_id.
3. Sort the totals descending and take the first 5 rows for the top-5 ranking.

Worked example

import pandas as pd

events = pd.DataFrame({
    "user_id":    [1, 2, 1, 3, 2, 1, 4, 5, 2, 3, 6, 1],
    "event_type": ["click", "click", "view", "purchase", "view", "purchase",
                    "click", "signup", "purchase", "view", "click", "signup"],
    "value":      [1, 1, 0, 50, 0, 30, 1, 0, 20, 0, 1, 0],
    "event_time": pd.date_range("2024-01-01", periods=12, freq="h"),
})

agg = events.groupby("event_type", as_index=False).agg(
    total_value=("value", "sum"),
    avg_value=("value", "mean"),
)
unique_users = events.groupby("event_type", as_index=False).agg(
    unique_users=("user_id", "nunique"),
)
top5 = agg.sort_values("total_value", ascending=False).head(5)

print(agg)
print(unique_users)
print(top5)

Output (verified against pandas 3.0.3):

  event_type  total_value  avg_value
0      click            4   1.000000
1   purchase          100  33.333333
2     signup            0   0.000000
3       view            0   0.000000
  event_type  unique_users
0      click             4
1   purchase             3
2     signup             2
3       view             3
  event_type  total_value  avg_value
1   purchase          100  33.333333
0      click            4   1.000000
2     signup            0   0.000000
3       view            0   0.000000

top5 shows all 4 event types here (fewer than 5 exist in this sample), correctly ordered by total_value descending.

Role of as_index and reset_index

• With as_index=False (used above), the group key (event_type) comes back as an ordinary column and the result already has a default RangeIndex, ready to sort, merge, or write out without an extra step.
• With the default as_index=True, the group key becomes the DataFrame's index instead of a column; call .reset_index() afterward to turn it back into a column, which is exactly what as_index=False does inline. The two are equivalent in the end result; as_index=False just skips the separate call.
• For a data-pipeline context specifically, returning tidy, index-free DataFrames at each stage (rather than index-heavy intermediate results) simplifies downstream joins and serialization, which is why as_index=False (or an explicit .reset_index()) is the more common choice in ETL (extract-transform-load) code even though as_index=True is pandas' own default.

Complexity

Each .groupby().agg() call is a single O(n) pass over events (n = number of rows), regardless of the number of distinct event_type values k; the result size is O(k). .nunique() per group additionally costs proportional to the number of rows in that group to track distinct values, still O(n) overall. sort_values().head(5) on the k-row aggregate is negligible next to the O(n) grouping pass, since k is typically tiny.

Edge cases

• An event_type with zero total value (as with signup and view above): still appears in the output with total_value=0 and avg_value=0.0, not silently dropped, since groupby includes every group with at least one row by default.
• Fewer than 5 distinct event types: .head(5) simply returns all of them, as shown; no error or special-casing needed.
• A value column containing NaN: .sum() and .mean() skip NaN by default (skipna=True), so a group with some missing values still gets a meaningful (if partial) aggregate rather than becoming NaN itself; be explicit in documentation if that silent skip is not the intended behavior for your data.

Trade-offs and pitfalls

• Named aggregation (total_value=('value','sum')) is preferred over the older dict-based .agg({'value': ['sum','mean']}) form specifically because the dict form produces MultiIndex columns (('value','sum'), ('value','mean')) that need flattening before they are convenient to use, while named aggregation gives flat, chosen names directly.

Direct answer

For a team of 20 analysts running occasional heavy ad-hoc queries, recommend the managed cloud OLAP (online analytical processing: query patterns dominated by large scans and aggregations, as opposed to OLTP, online transaction processing, which is dominated by many small transactional writes) service, BigQuery, over a self-managed Presto+Parquet cluster on Kubernetes. The deciding factor is not raw compute cost, where the two are close, but that a self-managed cluster is provisioned capacity billed whether it's busy or idle, and it needs a platform team to keep it healthy; at 20 analysts' actual query volume, that fixed cost overwhelms any per-query savings self-hosting offers.

Comparing the two on the five stated axes

Cloud pricing components to model, explicitly

A cost comparison that skips any of these components understates one side or the other:

• Compute: BigQuery's per-TB-scanned on-demand rate, or a flat-rate/editions commitment; self-managed compute-hours for the Kubernetes node pool.
• Storage: BigQuery's active logical storage rate; object storage (S3/GCS-class: services like Amazon S3 or Google Cloud Storage that store flat, key-addressed files rather than a traditional filesystem) for the Parquet files in the self-managed case.
• Network egress: cost of moving data out of the cloud region or between the storage layer and compute nodes, often ignored until a cross-region setup makes it material.
• Committed-use / reserved discounts: BigQuery editions/flat-rate slots, or reserved-instance pricing for the self-managed nodes; both can shift the comparison significantly at steady, predictable load.
• Support tier cost: enterprise support contracts, if the team needs guaranteed response times, are a real line item on either side that's easy to forget in a first-pass estimate.

Worked example

Verified current BigQuery rates (cloud.google.com/bigquery/pricing, checked 2026-09-24): on-demand analysis at $6.25 per TiB scanned (US multi-region), with the first 1 TiB/month free, and active logical storage at $0.02 per GiB-month. Model 20 analysts each running 15 queries/day, 22 working days/month, averaging 50 GiB scanned per query, against a 20 TiB resident dataset.

analysts, queries_per_day, days = 20, 15, 22
scan_per_query_GiB, dataset_GiB = 50, 20_000
queries_per_month = analysts * queries_per_day * days
scanned_TiB = queries_per_month * scan_per_query_GiB / 1024
print(f"queries/month: {queries_per_month:,}, scanned: {scanned_TiB:.1f} TiB")

bq_compute = max(0, scanned_TiB - 1) * 6.25       # VERIFIED rate, first 1 TiB free
bq_storage = dataset_GiB * 0.02                   # VERIFIED rate
print(f"BigQuery: ${bq_compute + bq_storage:,.2f}/mo (compute ${bq_compute:,.2f} + storage ${bq_storage:,.2f})")

# Self-managed: ASSUMPTION rates, pinned explicitly (no live quote for a specific node type)
nodes, vcpu_per_node, compute_rate = 10, 16, 0.045
storage_rate, fte, salary = 0.023, 1.5, 190_000
sm_compute = nodes * vcpu_per_node * compute_rate * 730
sm_storage = dataset_GiB * storage_rate
sm_staffing = fte * salary / 12
sm_total = sm_compute + sm_storage + sm_staffing
print(f"Self-managed: ${sm_total:,.2f}/mo (compute ${sm_compute:,.2f} + storage ${sm_storage:,.2f} + staffing ${sm_staffing:,.2f})")

queries/month: 6,600, scanned: 322.3 TiB
BigQuery: $2,407.91/mo (compute $2,007.91 + storage $400.00)
Self-managed: $29,466.00/mo (compute $5,256.00 + storage $460.00 + staffing $23,750.00)

At this team's actual query volume, BigQuery costs about 12x less than the self-managed cluster, almost entirely because the self-managed side pays for 1.5 platform engineers (an assumed, but realistic, staffing level to keep Presto and Kubernetes healthy) whether or not the cluster is busy. The break-even point, where BigQuery's scanned-data cost alone equals the self-managed cluster's fixed compute-plus-storage-plus-staffing cost, works out to roughly 4,700 TiB scanned per month in this model: about 15x this team's actual usage. Self-hosting would only become competitive if the team's query volume grew far beyond what 20 analysts running occasional heavy queries would plausibly generate, or if the platform team's time were already a sunk cost shared across other self-hosted systems.

Trade-offs and pitfalls

• The recommendation flips on scale, not on principle. Once sustained scanned-data volume is high enough (roughly 15x this team's usage, per the break-even above), the self-managed cluster's fixed cost starts to look cheap relative to BigQuery's linear per-scan billing.
• Vendor lock-in is real and worth naming, even when recommending BigQuery. If portability is a hard organizational requirement (regulatory, multi-cloud strategy), that can outweigh the cost argument; that's a legitimate reason to choose the more expensive, more portable option deliberately.
• BigQuery's on-demand billing rewards query discipline. An unindexed SELECT * over a huge table is expensive precisely because you pay per byte scanned; partitioning and clustering the underlying tables materially changes the cost side of this comparison and should be table stakes before concluding BigQuery is "too expensive."
• A self-managed estimate that omits Parquet file compaction and small-file cleanup will understate the real operational burden. Small-file proliferation from frequent writes is one of the most common sources of unplanned Presto-on-Parquet maintenance work.

Direct answer

Using the accounting (banker's) method, charge each append a constant amortized cost of $c = \frac{b}{b-1}$ for a growth factor b, and show that the credit this overcharge accumulates between resizes is always enough to pay for the next resize's copying cost; that gives a total cost of n appends bounded by $O(n)$, so $O(1)$ amortized per append. Redoing the same argument for $b = 1.5$ gives $c = 3$ instead of $c = 2$ (doubling's value): the amortized bound of $O(1)$ per append still holds for any fixed growth factor greater than 1, only the constant changes.

Structured elaboration

Setup

Assume, for the proof, that capacity grows through the sequence $1, b, b^2, \ldots$ (the standard simplifying assumption; real implementations round capacities to integers, which does not change the asymptotic result). A resize happens each time the array is full and one more element is appended: the m elements already present are copied into a new array of capacity $b \times m$, then the new element is inserted.

Accounting (banker's) method

Charge every append operation, resize or not, a fixed amortized cost $c = \frac{b}{b-1}$. An ordinary append (no resize) has actual cost 1, so it banks a credit of $c - 1 = \frac{1}{b-1}$ on the element just inserted. Between one resize (at capacity m) and the next (at capacity $bm$), exactly $(b-1)m$ ordinary appends happen before the array fills again, each banking $\frac{1}{b-1}$ in credit:
$(b-1)m \cdot \frac{1}{b-1} = m$
That accumulated credit, m, is exactly enough to pay the actual cost of copying the m existing elements at the next resize (the resize's own new element is paid for directly by that operation's own charge $c$). Since every operation's amortized charge is bounded by the constant c and no operation is ever left with a deficit, the total actual cost of n appends is at most $c \cdot n = \frac{b}{b-1} n$, which is $O(n)$, giving $O(1)$ amortized cost per append.

Aggregate-method cross-check

The same bound falls out of the aggregate method directly: resizes happen at capacities $1, b, b^2, \ldots, b^{K-1}$ for some K with $b^{K-1} < n \le b^K$, and the total number of elements ever copied across all resizes is the sum of those capacities:
$\begin{aligned} \text{TotalCopyCost} &= \sum_{k=0}^{K-1} b^k = \frac{b^K - 1}{b - 1} \\ &< \frac{b \cdot b^{K-1}}{b - 1} \le \frac{b}{b-1}\, n \end{aligned}$
Adding the n direct insertion costs, total cost is $O(n)$, matching the accounting-method bound exactly.

Redoing it for b = 1.5 versus b = 2

Plugging each growth factor into $c = \frac{b}{b-1}$:
$b = 2 \Rightarrow \frac{b}{b-1} = 2$
$b = 1.5 \Rightarrow \frac{b}{b-1} = 3$
Both are finite constants, so the amortized bound of $O(1)$ per append holds for either growth factor; what changes is only the constant itself, not the asymptotic class. Doubling copies each element roughly 2 times on average over its lifetime in the array; growth factor 1.5 copies each element roughly 3 times on average, in exchange for a smaller memory ceiling: at any moment, the wasted (allocated but unused) capacity is bounded by $(b-1)$ times the current size, so doubling can waste up to 100% of the array's used size while 1.5x growth wastes up to 50%.

Worked example

Trace the accounting method on a concrete sequence with $b = 2$: starting from capacity 1, resizes occur when appending the 2nd, 3rd, 5th, and 9th elements, at capacities 1, 2, 4, and 8 respectively. Copy costs at those four resizes are 1, 2, 4, and 8 elements respectively (summing the previous capacity each time), totaling $1+2+4+8=15$ copies for 9 appends, i.e., $15/9 \approx 1.67$ copies per append, comfortably under the proof's bound of $c - 1 = 1$ extra copy-credit per ordinary append plus the resize's own direct cost, and well under the $O(n)$ total the proof guarantees.

Trade-offs & pitfalls

The most common wrong turn is trying to bound the worst single append's cost rather than the total cost across all n appends: any individual resize genuinely costs $O(m)$, not $O(1)$, so a correct argument must be amortized (spread across the whole sequence) rather than claiming every append is cheap in isolation. A second common gap is treating "amortized $O(1)$" as meaning every append is fast in practice: a resize append is still a real, comparatively expensive operation the moment it happens (relevant for tail latency, even if the average over many appends is small), which matters if the system cares about worst-case per-operation latency rather than only total throughput. On the growth-factor trade-off itself, a common oversimplification is treating 1.5 as strictly better because it wastes less memory, without naming that it does so by paying for roughly 50% more total copying over the structure's lifetime; the right choice depends on whether the system is more memory-constrained or more copy-cost-sensitive.

Direct answer

An idempotent daily incremental-load DAG needs four things working together: staging-first writes that never touch the production table directly, an explicit run identifier scoping every write to exactly one logical attempt, deduplication before the final promotion, and a metadata record of what was actually committed. Together these mean retrying the same logical day, whether from a single task retry or a full DAG rerun, converges to the same final state instead of duplicating or losing rows.

Structured elaboration

Task granularity. Split the load into distinct tasks along natural idempotency boundaries: extract-to-staging, pre-checks, deduplicate-within-staging, atomic-swap-or-merge-into-target, and record-metadata. Splitting this finely, rather than one monolithic load task, means a failure at, say, the pre-check step can retry just that step without re-extracting, and it gives a clear per-step audit trail of exactly what happened on a given run.

Staging tables. The extract task writes to a staging table scoped by run identifier, either a uniquely named table per run or a shared staging table with a run_id column, never directly to the production table. A retried or even duplicated extract simply produces another (or overwrites the same) staging row set, with zero risk to the already-correct production table until the swap step deliberately promotes it.

Run identifiers. Every staged row carries an explicit run identifier, typically the DAG run's logical date combined with a monotonic attempt number, rather than relying on wall-clock insert time. This is what lets deduplication and metadata recording answer "which specific attempt produced this data" unambiguously, instead of inferring it from timing, which breaks under retries and backfills.

Pre-checks. Before promoting staged data, validate the row count against an expected historical range, validate that the expected columns are present, and validate that no key column has an unexpected null rate. A pre-check failure should halt before the swap, not attempt to promote suspect data and hope a downstream consumer catches it later.

Deduplication. Within staging, deduplicate on the row's natural business key, not a surrogate row identifier that would not survive a retried extract, keeping the most recent version when duplicates exist from overlapping incremental windows.

Atomic swap or MERGE. Promote staging into production either through an atomic rename or swap (the staging table becomes the new partition instantaneously) or a MERGE (or INSERT ... ON CONFLICT DO UPDATE) keyed on the business key, so a retry of this exact step, even after a partial completion, converges to the same end state rather than duplicating already-promoted rows.

Metadata recording. After a successful promotion, write a row to a metadata table (run_id, logical_date, row_count, status='committed', committed_at) that acts as the definitive record of what has actually been committed for that logical date. A subsequent run for the same date checks this metadata first, so an accidental double-trigger, two DAG runs for the same date somehow both firing, can detect the prior commit and skip redundant work rather than blindly reprocessing.

Transactional writes to the warehouse. Where the target warehouse supports it, wrap the final swap or merge and the metadata-recording insert in a single database transaction, so a crash between "data promoted" and "metadata recorded" cannot happen: either both commit, or neither does. This closes the small window that staging, run identifiers, and deduplication alone leave open, where data has landed in production but the metadata table does not yet know it, which could otherwise let a naive check-metadata-first design re-promote data that already landed.

Handling partial DAG failure explicitly. If the DAG fails partway, for example after the swap but before metadata recording, in a warehouse where the two cannot share one transaction, the retry logic must re-check the actual target-table state rather than re-run blindly. The metadata-recording task itself, on retry, should verify whether its own prior attempt already landed, an idempotent metadata write keyed on run_id such as INSERT ... ON CONFLICT DO NOTHING, rather than assuming a clean slate every time it runs.

Worked example

A daily incremental load of an orders table runs with run_id = 20260720_020000_attempt1. The extract lands 48,213 rows into staging_orders. The pre-check's expected row-count range is 40,000 to 60,000 based on historical daily volume, so 48,213 passes.

Deduplication finds 312 duplicate business keys from an overlapping incremental window with yesterday's late-arriving data, and keeps the most recent occurrence of each, leaving:

$48213 - 312 = 47901 \text{ unique rows}$

The MERGE into the orders production table affects all 47,901 rows, each either updating an existing row or inserting a new one depending on whether its business key already exists. The metadata table then records run_id=20260720_020000_attempt1, row_count=47901, status=committed.

Later that same day, the DAG is manually retried after a downstream consumer flags something and asks for a rerun. The metadata-check-first step queries the metadata table, finds logical_date=2026-07-20 already has a committed row with row_count=47901, and either skips straight to a no-op success if the retry was accidental, or, if the source data genuinely changed since the last run, proceeds through the same extract, deduplicate, merge, and metadata sequence again under a new run_id, landing whatever the currently correct row count is through the same idempotent MERGE, converging to the correct final state regardless of how many times it is repeated.

Trade-offs and pitfalls

Over-fragmenting task granularity, a separate task for every tiny sub-step, adds scheduling overhead and dependency-graph visual complexity that may not be worth it for a small pipeline; granularity should track genuine retry and audit boundaries, not be maximized for its own sake.

The metadata-check-first pattern only actually prevents double-processing if every code path that can trigger a run, including a human manually running the DAG from the orchestrator's own user interface, goes through it. A common gap is a manual "clear and rerun" action that bypasses the intended metadata check entirely.

Deduplication that blindly keeps the latest occurrence can silently mask a genuine data-quality bug: if a business key should never legitimately appear twice in a correct extract, always deduplicating instead of alerting on the first unexpected duplicate trades away a real correctness signal for a smoother-looking run.

Direct answer

A slow-drip silent failure, where row counts decay gradually over weeks rather than dropping suddenly, needs a detector tuned for TREND, not point-in-time deviation: comparing a smoothed recent trend against a smoothed historical trend, rather than comparing today's raw count to yesterday's or to a single historical baseline, since a gradual decline can hide inside normal day-to-day noise for a long time before any single day looks obviously wrong.

Structured elaboration

• Why point-in-time checks miss this: a simple "today vs. yesterday" or "today vs. trailing median" check tolerates small day-to-day swings by design, and a slow decay (losing 1-2% per week) stays within that normal-variance tolerance band for a long time, each individual day looks fine relative to its immediate neighbors even as the underlying trend is clearly downward over a longer window.
• Detector design: compute a smoothed trend line (a moving average over, say, 14 days) and compare its SLOPE over the trailing 4-8 weeks against a near-zero or seasonally-adjusted expected slope; a sustained negative slope over multiple weeks, even a small one, is the signal, not any single day's absolute value.
• Distinguishing decay from legitimate seasonal decline: overlay the same weekday comparison across YEARS, not just weeks, if this pattern recurs at the same calendar period every year (a known seasonal dip), it's legitimate; if the decline is unprecedented relative to the same period in prior years, it's more likely a real failure.
• Sensitivity tuning: set the slope-detection threshold based on the smallest decay rate you actually care about catching before it becomes a real problem (for example, a cumulative 10% loss over 8 weeks), rather than trying to catch every tiny fluctuation, which would be too noisy to be actionable.

Worked example

Concretely: a table's daily row count has a typical range of 95,000-105,000 with no obvious pattern day to day, so no point-in-time alert ever fires. Plotted as a 14-day moving average over 8 weeks, though, the trend line drops from about 102,000 to about 89,000, a roughly 13% decline with a clearly negative slope. The seasonal check confirms this same 8-week period last year showed a flat trend, ruling out a recurring seasonal explanation. Tracing which layer is responsible: producer logs, still emitting at a normal rate; Kafka retention/offsets, no gaps; consumer/ETL (extract, transform, load) layer, this is where the investigation lands, checking recent deploys to the consumer finds a filter condition added five weeks ago that was slightly too aggressive, silently excluding a small but growing subset of legitimate records as an edge case in the filter logic gradually widened its unintended impact.

Trade-offs and pitfalls

A slope-based trend detector inherently has a longer detection lag than a point-in-time anomaly detector, by design, since it needs several weeks of data to distinguish a real trend from noise, which is the trade-off you accept for catching a class of failure that point-in-time checks structurally cannot see. The pitfall is setting the trend window too short, a 5-day moving average is still noisy enough to false-positive on ordinary week-to-week variance, while a window long enough to be stable (14+ days) means a slow-drip failure that started only 10 days ago won't show a clear trend yet, so this detector complements rather than replaces point-in-time checks, it doesn't substitute for them.

Each of these three snippets fails for a different reason, and the fix generalizes into one robust pattern for exclusion lists supplied by an application.

Structured elaboration

1. WHERE status IN () is a syntax error in most engines (or, in dialects that allow it, matches nothing) because an empty list has no values to compare against. There is no single "corrected" form independent of intent: if the empty list means "no restriction was requested", drop the IN clause entirely before the query is built; if it means "this filter should legitimately match nothing" (e.g. a UI multi-select with every option unchecked), the honest, portable equivalent is WHERE FALSE, valid everywhere and behaving consistently instead of relying on how a particular engine happens to parse an empty list.
2. WHERE discount = NULL is not a syntax error but always evaluates to UNKNOWN (never TRUE), because NULL represents "unknown value", and no comparison operator, including =, can determine two unknowns are equal. Correct form: WHERE discount IS NULL.
3. WHERE order_date BETWEEN '2024-03-01' AND is an incomplete statement (missing the upper bound) and is a syntax error. Correct form, supplying the missing bound: WHERE order_date BETWEEN '2024-03-01' AND '2024-03-31'.

Generalizing to a parameterized exclusion list: an application variable :excluded_statuses might arrive as NULL (no filter intended) or as an empty array (filter to nothing). Two safe patterns:

-- NOT IN, guarded against NULL and empty
WHERE (:excluded_statuses IS NULL OR status NOT IN (SELECT UNNEST(:excluded_statuses)))
-- NOT EXISTS, the more robust default (see Trade-offs)
WHERE NOT EXISTS (
  SELECT 1 FROM UNNEST(:excluded_statuses) AS excl(status_val)
  WHERE excl.status_val = status
)

(UNNEST expands an array value into one row per element, so an array parameter can be compared against, or joined to, row by row, the same way a real table would be.) Checking for NULL up front in the NOT IN version means "no exclusions" doesn't accidentally collapse to NOT IN (), which some engines error on and others silently treat as "exclude nothing" or "exclude everything" depending on the engine, an inconsistency worth not relying on. The NOT EXISTS version doesn't need that guard at all: if :excluded_statuses is an empty array, UNNEST produces zero rows, NOT EXISTS over zero rows is always TRUE, and every order is correctly kept, no special-casing required.

The reason NOT EXISTS doesn't need the guard, and NOT IN does, is the NULL-poisoning problem itself: if :excluded_statuses contains a NULL element (say, a bad value slipped in from the application), status NOT IN (SELECT UNNEST(:excluded_statuses)) compares status against that NULL for every row, and status <> NULL evaluates to UNKNOWN. NOT IN's OR-chain of comparisons needs every single comparison to be definitively TRUE to include a row; one UNKNOWN in the chain poisons the whole thing to UNKNOWN (never TRUE), so the query silently returns ZERO rows, for every row, not just ones that would have matched the NULL. NOT EXISTS never has this problem, because it only ever asks "does at least one matching row exist", a question a NULL row can fail to answer TRUE to without poisoning anything else in the chain.

Worked example

Given orders(1,'shipped'), (2,'cancelled'), (3,'pending') and an exclusion list of ('cancelled'): the NOT EXISTS-based exclusion, WHERE NOT EXISTS (SELECT 1 FROM UNNEST(:excluded_statuses) AS excl(status_val) WHERE excl.status_val = status), correctly returns orders 1 and 3, excluding order 2 (status 'cancelled' matches the one excluded value).

With an EMPTY exclusion list (:excluded_statuses = []): UNNEST produces zero rows, so NOT EXISTS is TRUE for every order with no special-case needed, and the query correctly returns all three orders, 1, 2, and 3.

With a NULL slipped into the exclusion list (:excluded_statuses = ['cancelled', NULL]): NOT EXISTS still correctly returns orders 1 and 3, unaffected by the NULL element, since the NULL element simply never matches any row's status, it doesn't need to disprove anything for the other rows. Contrast this with the NOT IN version on the same NULL-containing list: status NOT IN ('cancelled', NULL) returns ZERO rows, silently dropping orders 1 and 3 too, not just excluding order 2, exactly the NULL-poisoning failure this whole pattern exists to avoid.

Trade-offs and pitfalls

The single most reliable fix across engines is NOT EXISTS rather than NOT IN, because NOT EXISTS never has the NULL-poisoning problem NOT IN has, demonstrated above: reach for it by default in exclusion-list logic. The NOT IN plus explicit NULL/empty-list guard shown above is still a legitimate pattern when a team already has NOT IN-based logic and wants the smallest possible diff, it just needs the guard NOT EXISTS gets for free.

Direct answer

For an organization with heavy regulatory audit requirements and frequent backfills, I'd lean Lambda, not Kappa, because auditors and backfills both want a batch layer that recomputes deterministically from the immutable historical record, which is exactly what Lambda's batch layer already is; asking a purely-streaming system to serve that role means your regulatory correctness story depends entirely on your streaming engine's replay guarantees, which is a riskier place to put it.

Structured elaboration

Operational complexity: Lambda genuinely costs more day to day (two systems to run, two deploy pipelines, two on-call surfaces), but for a regulated business, that cost buys you something auditors specifically want: a batch recomputation that is simple to explain, easy to freeze and re-run against a specific historical snapshot, and doesn't depend on the availability or configuration of a live streaming cluster at audit time.

Duplication of logic: this is Lambda's real weakness, and in a regulated environment it's worse than usual, because a divergence between the batch and speed layer isn't just an engineering embarrassment, it can be a compliance finding ("which number is the official one?"). The mitigation is to push as much shared logic as possible into a common library (or a framework like Apache Beam that can target both a batch and a streaming runner from one codebase), rather than accept two independent implementations.

Storage costs: both approaches need to retain full history for backfills either way (Kappa needs it for replay, Lambda's batch layer needs it for recompute), so this axis is close to a wash; Lambda's marginal storage cost is really just the speed layer's short-lived recent-window state, which is small.

Correctness guarantees: this is where Lambda wins for this specific org. The batch layer is the audit-facing source of truth, computed the same deterministic way every time from an immutable log, and the speed layer is explicitly labeled as "provisional, not yet reconciled," which is a distinction regulators understand. A backfill in Lambda just means rerunning the batch layer over a corrected range; in Kappa it means replaying the single streaming codebase, which, if that codebase has since changed, can produce a subtly different result for old data than what actually shipped at the time, which is a genuinely uncomfortable position to be in during an audit.

Worked example

Concretely: this company discovers in Q3 that a transaction was miscategorized in Q1. Under Lambda, the fix is: correct the source record, rerun the Q1 batch job, republish the corrected view, and the batch layer's determinism means the new number is defensible on its own. Under Kappa, the fix requires replaying the stream from Q1 through today's version of the processing code, and now you have to prove to an auditor that today's code, run against Q1 data, produces the same result the business actually reported in Q1, which is a much harder claim to substantiate if the code has evolved.

Trade-offs and pitfalls

A small team with limited SRE resources changes this calculus, but not the recommendation, only the mitigation: they should still lean Lambda for the audit story, but invest hard in sharing logic (Beam, or a well-tested shared library called from both layers) rather than accepting two hand-maintained pipelines, because a small team maintaining two fully independent codepaths is where Lambda's operational cost actually becomes unsustainable. The mistake to avoid is picking Kappa purely for the team's operational convenience while ignoring that it shifts audit risk onto code-evolution guarantees the team may not actually be able to make.

Window functions are evaluated after WHERE filters rows, so a window function's result doesn't exist yet at the point WHERE runs; SQL simply doesn't allow referencing one there (or in GROUP BY, for the same reason). To filter on a cumulative spend computed with a window function, compute it first in a CTE (common table expression, a named WITH block) or subquery, then filter the outer query against that already-computed column.

Approach

WITH customer_year_spend AS (
  SELECT
    customer_id,
    order_id,
    order_date,
    amount,
    SUM(amount) OVER (
      PARTITION BY customer_id, EXTRACT(YEAR FROM order_date)
      ORDER BY order_date, order_id
      ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS cumulative_spend
  FROM sales
)
SELECT DISTINCT customer_id
FROM customer_year_spend
WHERE cumulative_spend > 1000;

Key points

• The window function runs inside the CTE, over the full row set; the outer query's WHERE filters the CTE's already-materialized output, which is a normal column at that point, not a window function call.
• The explicit ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW frame matters: without it, the default frame for an ORDER BY-bearing window is RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, which groups every row sharing the same ORDER BY value (here, the same order_date) into one combined total for all of them, not a true row-by-row running sum.
• Because amount is non-negative, the cumulative sum only ever increases through the year, so "the running total ever exceeds 1000" and "the year-end total exceeds 1000" pick out the exact same set of customers; DISTINCT customer_id over any row that clears the threshold is equivalent to filtering on the final total.

Why ROWS matters here: the RANGE default-frame trap

Two orders for the same customer on the same order_date (200 on 2024-03-10 and 150 on 2024-03-10, following an earlier 100 on 2024-01-05) expose the bug directly. With the default RANGE frame, both same-day rows are treated as peers and both get the combined total of all peers as their cumulative value (450 for both), instead of the running total accumulating row by row (300 after the first of the two, 450 after the second). ROWS fixes this because it counts physical rows up to the current one, not "all rows with the same ORDER BY value."

(Verified by executing both the ROWS-framed and RANGE-framed versions of this window against the same data in SQLite 3.51; the RANGE result matches the buggy pattern exactly as shown.)

Complexity

Computing the window function is one ordered pass per partition: O(n log n) if the engine has to sort by (customer_id, order_date, order_id), or close to O(n) if an existing index already provides that order. The outer WHERE/DISTINCT is a single O(n) filter and de-duplication pass over the CTE's output. This is the same total cost as computing a plain GROUP BY/HAVING total would be; the window function version costs more only in the sense that it materializes a row per order instead of collapsing straight to one row per customer.

Edge cases

• Same-order_date orders: handled correctly only with the explicit ROWS frame, as shown above; this is the single most common way this pattern silently breaks.
• Negative amounts (refunds): if amount can go negative, the cumulative sum is no longer monotonically increasing, and "any row exceeds the threshold" is no longer equivalent to "the year-end total exceeds the threshold"; in that case you must filter on the maximum cumulative value per customer-year, not just any exceeding row, or decide explicitly which semantics the business actually wants.
• Customers with zero orders in the year: they never appear in sales at all for that year, so they're correctly absent from the result without any extra handling.

Trade-offs and pitfalls

If the business question is genuinely just "whose total for the year exceeds the threshold," a plain GROUP BY customer_id HAVING SUM(amount) > 1000 computes the same customer set with less work: one aggregation pass, no window function, no CTE wrapper, and no risk of the RANGE/ROWS frame trap at all. The window-function-plus-CTE form earns its keep when the question is really about the running value, not just the total: for example, flagging the exact order at which a customer crossed a loyalty threshold, not just whether they crossed it by year-end. Reach for GROUP BY/HAVING when you only need the final number, and reach for the windowed CTE when the row-by-row trajectory itself is part of what's being asked.

Direct answer

Storing created_at as TIMESTAMP WITHOUT TIME ZONE means the column silently records whatever local time the writer happened to be in, with no record of which timezone that was; the fix is to change the column to TIMESTAMPTZ and normalize all writes to UTC at ingestion, not to try to reconstruct timezone information after the fact from a column that never captured it.

Structured elaboration

• The risk across regions: if application servers in different regions write to this column using their own local clock (or worse, a mix of local time and UTC depending on which service wrote the row), two rows with the identical stored value can represent two genuinely different real-world instants, and there is no way to tell which is which after the fact, because the timezone information was never captured.
• Normalizing for analysis: the only fully correct fix is to know, for each historical row, what timezone or UTC offset it was actually written in; if that's recoverable (e.g., from a separate region column on the same table, or from deployment history correlated with created_at's range), you can backfill a corrected UTC value. If it's not recoverable, the historical data has an irreducible ambiguity that should be documented, not silently "fixed" with a guessed offset.
• Schema change vs. enforce-at-ingestion: the schema itself should change to TIMESTAMPTZ (which in Postgres always stores UTC internally and converts to/from a timezone only at display time), and the application layer must be fixed to always write in UTC (or with an explicit, correct offset) going forward; changing only the column type without fixing the writers just moves the same ambiguity into a differently-typed column.

Worked example

-- schema fix, step 1: correct any KNOWN non-UTC subset FIRST, while the column is still naive,
-- so the later blanket conversion doesn't reinterpret it a second time.
-- Reinterpret the naive value as the writer's actual local time, then re-express it as the
-- equivalent naive UTC wall-clock value (this two-step "AT TIME ZONE" is what actually corrects it):
UPDATE events
SET created_at = (created_at AT TIME ZONE 'America/Los_Angeles') AT TIME ZONE 'UTC'
WHERE region = 'us-west' AND created_at < '2026-03-01';

-- schema fix, step 2: now that every row's naive value is genuinely UTC, do the blanket type change
ALTER TABLE events ALTER COLUMN created_at TYPE TIMESTAMPTZ USING created_at AT TIME ZONE 'UTC';

The USING clause is doing real work here: TIMESTAMP WITHOUT TIME ZONE AT TIME ZONE 'UTC' reinterprets the naive timestamp as if it were UTC and produces a correct TIMESTAMPTZ; if the true original offset was NOT UTC for some rows, that reinterpretation is silently wrong for exactly those rows, which is why any known region-specific writer needs its own explicit correction pass, run BEFORE the blanket conversion, not after. Running the targeted correction after the column is already TIMESTAMPTZ does not fix it: confirmed by executing both orderings, converting a naive '2026-02-15 10:00:00' value that was actually written in America/Los_Angeles (true UTC instant 2026-02-15 18:00:00+00) using the blanket-first, correct-second order in this answer's original text produced 2026-02-15 02:00:00+00, which is wrong in the opposite direction; running the correction first, as shown above, produced the correct 2026-02-15 18:00:00+00.

Trade-offs and pitfalls

• The single most dangerous mistake here is applying one blanket AT TIME ZONE conversion to the whole table when the true originating offset actually varied by region or by writer over time; that "fixes" the type but silently bakes in a wrong absolute time for every row that wasn't actually UTC to begin with.
• A closely related mistake is getting the ORDER of operations backwards: any known region-specific correction must be applied to the still-naive column before the blanket ALTER ... USING ... AT TIME ZONE 'UTC' runs, not after. Once the blanket conversion has already relabeled every naive value as UTC, a single further AT TIME ZONE step applied to the resulting TIMESTAMPTZ does not correctly undo and reapply the right offset; it needs an explicit reversal step, so it is simpler and less error-prone to fix known-wrong subsets first, while they're still naive, and only then run the one blanket conversion.
• If the true originating offset for some historical rows is genuinely unrecoverable, converting them anyway to satisfy the new column type without flagging the uncertainty produces confidently-wrong data, which is worse than leaving a documented gap; a data_quality_note or a separate timezone_confidence flag on the affected range is more honest.
• Going forward, enforcing UTC at the application/ingestion layer (not just at the database schema level) is what actually prevents recurrence; a schema type change alone doesn't stop a misconfigured writer from producing a naive local timestamp that then gets miscast as if it were UTC.