Lineage and Pipeline Observability

## Why a green run proves almost nothing A task's terminal state answers one question: did the process finish without raising. Every interesting data failure passes that test. A source system exports a partial file; a filter predicate now excludes most rows because an upstream enum value changed; a join loses rows because a key column started arriving with different casing; an incremental load reads the wrong window and finds almost nothing. In every case the code runs perfectly and the pipeline goes green. This is the gap that **pipeline observability** exists to close, and it is why "is the pipeline healthy?" and "is the data correct?" are separate questions with separate instrumentation. ## The metrics worth recording on every run Record these as structured run metadata, not as log lines someone greps later: - **Rows in and rows out** per input and output dataset, and the **ratio** between them. A join or filter has a characteristic yield; a change in yield is often the earliest signal. - **Bytes or files written**, and the **partitions touched** — useful when a run writes to the wrong partition entirely. - **Duration**, and where possible duration per stage. A run that finishes in a tenth of the usual time processed a tenth of the usual work. - **Column-level facts** on a handful of important columns: null rate, distinct count, min and max of the event-time column, sum of a key measure. - **Freshness**: the maximum event timestamp present in the output, and the wall-clock age of the last successful production. The last two are what let you answer "is today's data sane?" rather than "did the job run?". ## Turning a number into a signal Raw counts are not alerts. You need a comparison: - **Against history** — the trailing median or mean of the same metric over recent comparable runs. Day-of-week matters: Sunday volume is legitimately a fraction of Tuesday volume, so compare like with like or the check will cry wolf every weekend and be muted. - **Against a business floor** — an absolute lower bound where one genuinely exists, for example a subscription table that must never shrink. - **Against the input** — an output/input ratio band is often more stable than either number alone, because it survives genuine volume swings. For the scenario in the question, the ratio check and the trailing-median check both fire immediately: one-tenth of the usual output is far outside any sane band. A more insidious version — 8% of rows silently dropped — is caught only if the band is tight enough, which is why teams tier checks: hard thresholds on datasets that feed billing or regulators, looser ones elsewhere. ```sql -- volume check against the trailing median for the same weekday with history as ( select run_date, rows_written from pipeline_run_metrics where dataset = 'orders_daily' and extract(dow from run_date) = extract(dow from current_date) and run_date >= current_date - interval '56 days' ) select (select rows_written from today) < 0.6 * (select percentile_cont(0.5) within group (order by rows_written) from history) as volume_alert; ``` ## Freshness is the metric consumers actually feel A row-count check on a run that *did not happen* never fires, because there is no run to check. Freshness is measured on the dataset, not on the run, so it catches skipped schedules, permanently stuck sensors and silently disabled pipelines. Two flavours are worth separating: **data freshness** (how old the newest record is) and **pipeline freshness** (how long since the dataset was last successfully written). A dataset written on time but containing only stale records fails the first and passes the second — usually pointing at a broken source rather than a broken pipeline. ## Where the check belongs A metric on a dashboard that nobody watches is not observability. Put the assertion inside the pipeline, as a task that runs after the write and before anything downstream consumes the result, so that a violated threshold fails the run and leaves downstream work unstarted. Emit the measured values as run metadata regardless of pass or fail, so you accumulate the history that makes tomorrow's baseline meaningful. ## Diagnosing this particular incident With per-run metrics in place, the walk is short. Compare rows in against rows out for the failing run: if input was normal and output collapsed, the loss is inside the transformation — check the filter, the join keys, the window used by the incremental predicate. If input also collapsed, walk upstream via lineage to the producing run and repeat. If input looks normal but the *distinct count of the join key* dropped, you have a key-format change. Every one of those steps needs numbers you only have if you recorded them at the time — reconstructing them after the fact is usually impossible once the table has been overwritten.