Skip to main content
Data Engineering · lesson 19 of 23

Beginner

8

Intermediate

4

Advanced

11

Also part of Data Engineering

9
Open the Data Engineering page →
Data Engineering advanced Lesson 19 of 23

Also part of: Monitoring & Observability

Pipeline Observability: Tracing, Anomaly Detection, and Alerting

Trace a pipeline run stage-by-stage, then flag a run 3x longer than usual and an error rate crossing its threshold - operational health, not data quality.

By SkillByExample · Published

data-engineering observability monitoring alerting

Trace one pipeline run stage by stage, then scan six runs of history for a duration anomaly and a rising error rate.

Run history: durations by stage, and outcome

run_history = [
    {"run_id": 501, "stages": {"extract": 12.1, "transform": 34.8, "load": 8.2}, "rows": 48200, "status": "success"},
    {"run_id": 502, "stages": {"extract": 11.4, "transform": 33.0, "load": 7.9}, "rows": 47100, "status": "success"},
    {"run_id": 503, "stages": {"extract": 13.0, "transform": 35.5, "load": 8.5}, "rows": 49020, "status": "success"},
    {"run_id": 504, "stages": {"extract": 45.2, "transform": 128.9, "load": 9.1}, "rows": 48688, "status": "success"},
    {"run_id": 505, "stages": {"extract": 12.6, "transform": 0.0, "load": 0.0}, "rows": 0, "status": "failed"},
    {"run_id": 506, "stages": {"extract": 12.9, "transform": 34.1, "load": 8.0}, "rows": 48810, "status": "success"},
]
for r in run_history:
    r["duration_s"] = sum(r["stages"].values())

Tracing one run stage by stage

def trace(run):
    print(f"run {run['run_id']} ({run['status']}), total {run['duration_s']:.1f}s:")
    for stage, seconds in run["stages"].items():
        pct = 100 * seconds / run["duration_s"] if run["duration_s"] else 0
        print(f"  {stage:<10} {seconds:>6.1f}s  ({pct:5.1f}%)")

trace(run_history[0])
$ python observability.py
== single-run trace: stage-by-stage breakdown ==
run 501 (success), total 55.1s:
  extract      12.1s  ( 22.0%)
  transform    34.8s  ( 63.2%)
  load          8.2s  ( 14.9%)

The trace immediately says where the time goes — transform is 63% of this run’s duration. When a run gets slower, this is what turns “the pipeline is slow” into “the transform stage is slow,” which is a much smaller place to start looking.

Duration anomaly: trailing average, not a fixed threshold

def check_duration_anomaly(history, threshold_multiplier=2.0):
    successful = [r for r in history if r["status"] == "success"]
    for i, run in enumerate(successful):
        prior = successful[:i]
        if len(prior) < 3:
            continue
        trailing_avg = sum(p["duration_s"] for p in prior[-3:]) / 3
        if run["duration_s"] > trailing_avg * threshold_multiplier:
            print(f"  ALERT run {run['run_id']}: {run['duration_s']:.1f}s vs trailing avg "
                  f"{trailing_avg:.1f}s ({run['duration_s'] / trailing_avg:.1f}x)")

check_duration_anomaly(run_history)
== duration anomaly detection: trailing average over successful runs ==
  ALERT run 504: 183.2s vs trailing avg 54.8s (3.3x)

Run 504 took 3.3x its own trailing average — not compared to some hardcoded “60 seconds is too slow” threshold, but to what this pipeline’s recent runs actually looked like. That means the same alerting code keeps working as the pipeline’s normal volume (and normal duration) grows over months, with no threshold to manually retune.

Error rate over a rolling window

last_5 = run_history[-5:]
error_rate = sum(1 for r in last_5 if r["status"] == "failed") / len(last_5)
print(f"  {sum(1 for r in last_5 if r['status'] == 'failed')}/{len(last_5)} failed = {error_rate:.0%}")
if error_rate > 0.15:
    print(f"  ALERT: error rate {error_rate:.0%} exceeds 15% threshold")
== error rate over the last 5 runs ==
  1/5 failed = 20%
  ALERT: error rate 20% exceeds 15% threshold

A single failed run (505) is one data point; whether it’s noise or a trend depends on the window around it. A 20% error rate over the last five runs crosses a 15% threshold worth paging on — a single failure buried in five hundred successful runs would not.

Practice

1. Add a run where the `load` stage alone takes 90 seconds, and re-trace it. 
run 507 (success), total 102.0s:
  extract       5.0s  (  4.9%)
  transform     7.0s  (  6.9%)
  load         90.0s  ( 88.2%)

The trace immediately isolates it — 88% of this run’s time is load, the opposite profile from run 501’s transform-heavy trace, and a different investigation entirely.

2. Lower the anomaly threshold multiplier from 2.0x to 1.5x and see what changes. 
  ALERT run 504: 183.2s vs trailing avg 54.8s (3.3x)

No new alerts for this data — run 504 is the only one far enough above trailing average to trip either threshold. A more sensitive threshold matters more on noisier real-world duration data than this clean example has.

3. Compute the error rate over the last 3 runs instead of the last 5. 
last_3 = run_history[-3:]
rate = sum(1 for r in last_3 if r["status"] == "failed") / len(last_3)
print(f"{rate:.0%}")
33%

Run 505 (the only failure) is still inside the last-3 window (runs 504, 505, 506), so this narrower window actually reports a higher rate — 33% vs. 20% — for the same single failure, simply because the denominator shrank from 5 to 3. Window size changes the reported rate even when nothing about the underlying failures changes, which is worth knowing before picking one.

4. Why does `check_duration_anomaly` skip the first 3 successful runs entirely?

if len(prior) < 3: continue — there’s no trailing average to compare against until at least 3 prior successful runs exist. Alerting against an average of zero or one data point would be noise, not signal; the function correctly declines to guess.

Next: cost optimization — where a pipeline’s compute and storage spend actually goes, measured, not guessed.

Frequently Asked Questions

How is observability different from the data-quality lesson's freshness checks?
Data quality asks whether the data itself is correct and complete — right row counts, valid values, on-time arrival. Observability asks whether the pipeline as a system is healthy — how long each run takes, which stage is slow, whether the failure rate is climbing. A pipeline can be perfectly observable (every run traced, every metric logged) while still producing bad data, and vice versa — they catch different failure classes.
What is a stage-by-stage trace, and why not just log total duration?
A trace breaks one run's duration into its component stages (extract, transform, load), so 'the pipeline got slower' becomes 'the transform stage got slower' — pointing straight at what to investigate instead of starting a blind search across the whole codebase.
Why compare against a trailing average instead of a fixed duration threshold?
A fixed threshold ("alert if over 60 seconds") has to be re-tuned every time data volume grows, and legitimately larger runs trigger false alarms. A trailing average adapts to the pipeline's actual recent behavior, so the alert fires on a real deviation from what's normal *now*, not from what was normal six months ago.
What should trigger a page vs. just a log entry?
A rule of thumb: alert on symptoms a human needs to act on soon (a duration spike that will miss an SLA, an error rate crossing a threshold), and just log everything else for later investigation. Paging on every metric trains people to ignore pages; logging everything but alerting on nothing means no one finds out until a downstream consumer complains.