It's 8:47 AM on a Tuesday and your VP of Sales is presenting the weekly revenue dashboard to the executive team. Halfway through the meeting, someone notices that the numbers haven't moved since Friday. Your Slack lights up. The pipeline ran — the orchestrator shows green — but the data is three days stale because an upstream API started throttling requests and your pipeline silently swallowed the partial load without complaint. No alert fired. No SLA was breached, technically, because you never formally defined one.
This is the gap that data pipeline SLAs are designed to close. Not the gap between "did the pipeline run?" and "did the pipeline fail?" — your orchestrator already handles that — but the gap between technical success and business value. A pipeline that runs on schedule but produces stale, incomplete, or late data has failed in the way that actually matters. SLAs give you the vocabulary, the measurements, and the enforcement mechanisms to catch that kind of failure before your VP's presentation, not during it.
By the end of this lesson, you'll be able to define precise SLA contracts for your data pipelines, instrument your pipelines to measure against those contracts in production, build alerting logic that catches real failures rather than just job crashes, and establish a review cadence that keeps SLAs relevant as your systems evolve.
What you'll learn:
You should already be comfortable with:
Before building anything, let's be precise about terminology, because the data industry has a habit of using "SLA," "SLO," and "SLI" interchangeably in ways that cause real operational confusion.
Borrowing from Google's Site Reliability Engineering framework:
In practice, most data teams conflate these, and that's fine as long as the person writing the pipeline and the person relying on its output share the same understanding. What matters is that your "SLA" answer four questions:
Without answers to all four, you don't have an SLA — you have a wish.
Most pipelines need SLAs across four dimensions. Understanding each one helps you choose the right measurements.
Freshness measures how recent the data in a target system is relative to some reference point, usually either wall clock time or the timestamp of the most recent source event.
A freshness SLA might read: "The orders table in the analytics warehouse should reflect events that occurred no more than 4 hours ago, measured at any point during business hours."
This is different from a latency SLA. Freshness is a property of the data at rest; latency is a property of the pipeline's processing speed. You can have fresh data delivered with high latency (if source data is slow to arrive) and you can have stale data despite low-latency pipelines (if the pipeline is failing silently).
Latency measures the time between when data enters the pipeline and when it becomes available to consumers downstream. It's typically measured as end-to-end pipeline duration, but you can also measure it per stage if you need to diagnose bottlenecks.
A latency SLA might read: "95% of CDC events captured from the production PostgreSQL database should be reflected in the data warehouse within 15 minutes of being committed to the source."
Completeness measures whether all expected data arrived. This is subtle: a pipeline can run successfully, finish on time, and produce fresh-looking data while silently dropping 30% of records due to a schema mismatch or API pagination bug.
A completeness SLA might read: "The daily revenue reconciliation pipeline should produce a row count within ±2% of the prior 7-day average. Deviations beyond ±5% should trigger an immediate alert."
Availability measures what fraction of the time your pipeline produces usable output. This is most relevant for pipelines that feed real-time dashboards or operational systems.
An availability SLA might read: "The customer-facing analytics endpoint should return data no older than 1 hour for 99% of requests between 6 AM and 10 PM UTC."
Why all four matter: A common mistake is to monitor only latency (because it's easy to measure from job logs) while ignoring freshness and completeness. The highest-severity incidents at most data teams involve completeness failures — not job crashes.
SLA design is a negotiation between what the business needs and what the system can reliably deliver. Here's how to structure that conversation.
The most common mistake in SLA design is starting with what the pipeline currently does and working backward to a threshold. That approach bakes in existing limitations as permanent constraints.
Instead, start by asking the data consumer: "What's the worst staleness you could tolerate before the data becomes misleading or harmful?" For a fraud detection model, the answer might be 5 minutes. For a quarterly board report, it might be 24 hours.
Once you have the consumer requirement, work backward through your pipeline architecture to identify whether you can actually meet it, and at what cost.
Not every pipeline deserves the same SLA rigor. Establish tiers:
| Tier | Description | Example | Typical Freshness SLA |
|---|---|---|---|
| P0 | Business-critical, customer-facing | Payment processing feed | < 5 minutes |
| P1 | Executive/operational reporting | Daily revenue dashboard | < 1 hour |
| P2 | Internal analytics | Marketing attribution model | < 4 hours |
| P3 | Data science feature engineering | Historical training datasets | < 24 hours |
The tier determines how aggressively you monitor, how quickly you escalate, and how much engineering investment you make in reliability.
Don't leave SLA definitions in a Confluence page that goes stale. Encode them in a configuration file that your monitoring code actually reads.
# pipeline_slas.yaml
pipelines:
orders_cdc_to_warehouse:
tier: P1
description: "CDC replication from orders PostgreSQL to BigQuery"
owner: "data-engineering@company.com"
escalation_channel: "#data-alerts-urgent"
slas:
freshness:
max_lag_minutes: 60
evaluation_window: "business_hours" # 6am-10pm UTC
measurement: "max(event_timestamp) in target vs wall_clock"
latency:
p95_minutes: 15
p99_minutes: 30
window_hours: 24
completeness:
row_count_tolerance_pct: 5.0
comparison_window_days: 7
daily_revenue_reconciliation:
tier: P1
description: "Nightly roll-up of revenue by product and region"
owner: "data-engineering@company.com"
escalation_channel: "#data-alerts-urgent"
slas:
freshness:
max_lag_hours: 2
evaluation_time: "08:00 UTC"
measurement: "pipeline completion timestamp vs scheduled time"
completeness:
row_count_tolerance_pct: 2.0
comparison_window_days: 14
minimum_rows: 10000
This YAML becomes the ground truth. Your monitoring system reads it; your on-call runbook references it; your SLA review process updates it.
You can't measure what you don't capture. Most orchestrators give you job-level start/end times, but SLA measurement requires deeper telemetry.
The foundation of SLA monitoring is a persistent audit table that every pipeline writes to. Here's a schema that handles the four dimensions we defined:
CREATE TABLE pipeline_audit (
run_id VARCHAR(64) NOT NULL,
pipeline_name VARCHAR(255) NOT NULL,
pipeline_version VARCHAR(32),
triggered_at TIMESTAMPTZ NOT NULL,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
status VARCHAR(32) NOT NULL, -- 'running', 'success', 'failed', 'partial'
records_read BIGINT,
records_written BIGINT,
records_failed BIGINT,
max_source_ts TIMESTAMPTZ, -- freshest event timestamp in source batch
min_source_ts TIMESTAMPTZ, -- oldest event timestamp in source batch
target_table VARCHAR(255),
error_message TEXT,
metadata JSONB, -- pipeline-specific key/value pairs
PRIMARY KEY (run_id)
);
CREATE INDEX idx_audit_pipeline_name ON pipeline_audit(pipeline_name, completed_at DESC);
CREATE INDEX idx_audit_status ON pipeline_audit(status, completed_at DESC);
Notice the status column includes 'partial' — this is critical. A pipeline that wrote 60% of expected records and exited cleanly is not a success. You need a status value that captures this condition.
Rather than manually writing audit records in every pipeline, wrap the logic in a context manager that handles both success and failure cases:
import uuid
import json
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Optional
import psycopg2
class PipelineAuditContext:
"""
Context manager for capturing pipeline run metadata to the audit table.
Usage:
with PipelineAuditContext("orders_cdc_to_warehouse", conn) as audit:
records = fetch_from_source()
audit.set_records_read(len(records))
written = load_to_warehouse(records)
audit.set_records_written(written)
audit.set_max_source_ts(records[-1]["event_timestamp"])
"""
def __init__(
self,
pipeline_name: str,
db_conn,
pipeline_version: str = "1.0.0",
target_table: Optional[str] = None,
triggered_at: Optional[datetime] = None,
):
self.pipeline_name = pipeline_name
self.db_conn = db_conn
self.pipeline_version = pipeline_version
self.target_table = target_table
self.run_id = str(uuid.uuid4())
self.triggered_at = triggered_at or datetime.now(timezone.utc)
self.started_at: Optional[datetime] = None
self.completed_at: Optional[datetime] = None
self.status = "running"
self.records_read: Optional[int] = None
self.records_written: Optional[int] = None
self.records_failed: Optional[int] = None
self.max_source_ts: Optional[datetime] = None
self.min_source_ts: Optional[datetime] = None
self.error_message: Optional[str] = None
self.metadata: dict = {}
def set_records_read(self, count: int):
self.records_read = count
def set_records_written(self, count: int):
self.records_written = count
def set_records_failed(self, count: int):
self.records_failed = count
def set_max_source_ts(self, ts: datetime):
self.max_source_ts = ts
def set_min_source_ts(self, ts: datetime):
self.min_source_ts = ts
def add_metadata(self, key: str, value):
self.metadata[key] = value
def _determine_final_status(self) -> str:
"""
Determine whether a technically successful run actually delivered
complete data. This is where silent failures get caught.
"""
if self.status == "failed":
return "failed"
if self.records_read is not None and self.records_written is not None:
if self.records_read > 0:
write_ratio = self.records_written / self.records_read
if write_ratio < 0.95:
# More than 5% of records were dropped
return "partial"
if self.records_failed is not None and self.records_failed > 0:
total = (self.records_written or 0) + self.records_failed
if total > 0 and self.records_failed / total > 0.05:
return "partial"
return "success"
def _write_audit_record(self):
cursor = self.db_conn.cursor()
cursor.execute(
"""
INSERT INTO pipeline_audit (
run_id, pipeline_name, pipeline_version, triggered_at,
started_at, completed_at, status, records_read,
records_written, records_failed, max_source_ts,
min_source_ts, target_table, error_message, metadata
) VALUES (
%(run_id)s, %(pipeline_name)s, %(pipeline_version)s,
%(triggered_at)s, %(started_at)s, %(completed_at)s,
%(status)s, %(records_read)s, %(records_written)s,
%(records_failed)s, %(max_source_ts)s, %(min_source_ts)s,
%(target_table)s, %(error_message)s, %(metadata)s
)
ON CONFLICT (run_id) DO UPDATE SET
completed_at = EXCLUDED.completed_at,
status = EXCLUDED.status,
records_read = EXCLUDED.records_read,
records_written = EXCLUDED.records_written,
records_failed = EXCLUDED.records_failed,
max_source_ts = EXCLUDED.max_source_ts,
error_message = EXCLUDED.error_message,
metadata = EXCLUDED.metadata
""",
{
"run_id": self.run_id,
"pipeline_name": self.pipeline_name,
"pipeline_version": self.pipeline_version,
"triggered_at": self.triggered_at,
"started_at": self.started_at,
"completed_at": self.completed_at,
"status": self.status,
"records_read": self.records_read,
"records_written": self.records_written,
"records_failed": self.records_failed,
"max_source_ts": self.max_source_ts,
"min_source_ts": self.min_source_ts,
"target_table": self.target_table,
"error_message": self.error_message,
"metadata": json.dumps(self.metadata),
},
)
self.db_conn.commit()
cursor.close()
def __enter__(self):
self.started_at = datetime.now(timezone.utc)
self.status = "running"
self._write_audit_record() # Write initial record so we can track stuck pipelines
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.completed_at = datetime.now(timezone.utc)
if exc_type is not None:
self.status = "failed"
self.error_message = f"{exc_type.__name__}: {str(exc_val)}"
else:
self.status = self._determine_final_status()
self._write_audit_record()
return False # Don't suppress exceptions
With this in place, every pipeline run — whether it succeeds, fails, or produces partial output — leaves a complete, queryable record.
The audit table is your raw material. Now you need a monitoring process that reads it and evaluates it against your SLA definitions.
import yaml
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass
from typing import List, Optional
import psycopg2.extras
@dataclass
class SLAViolation:
pipeline_name: str
tier: str
violation_type: str # 'freshness', 'latency', 'completeness', 'availability'
description: str
severity: str # 'warning', 'critical'
measured_value: float
threshold_value: float
run_id: Optional[str] = None
detected_at: datetime = None
def __post_init__(self):
if self.detected_at is None:
self.detected_at = datetime.now(timezone.utc)
class SLAEvaluator:
def __init__(self, sla_config_path: str, db_conn):
with open(sla_config_path) as f:
self.config = yaml.safe_load(f)
self.db_conn = db_conn
def evaluate_all(self) -> List[SLAViolation]:
violations = []
for pipeline_name, pipeline_config in self.config["pipelines"].items():
violations.extend(self.evaluate_pipeline(pipeline_name, pipeline_config))
return violations
def evaluate_pipeline(
self, pipeline_name: str, config: dict
) -> List[SLAViolation]:
violations = []
slas = config.get("slas", {})
if "freshness" in slas:
v = self._check_freshness(pipeline_name, config, slas["freshness"])
if v:
violations.append(v)
if "latency" in slas:
violations.extend(
self._check_latency(pipeline_name, config, slas["latency"])
)
if "completeness" in slas:
v = self._check_completeness(pipeline_name, config, slas["completeness"])
if v:
violations.append(v)
return violations
def _check_freshness(
self, pipeline_name: str, config: dict, freshness_sla: dict
) -> Optional[SLAViolation]:
"""
Check if the most recent successful run's max_source_ts is within
the freshness threshold.
"""
cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cursor.execute(
"""
SELECT run_id, completed_at, max_source_ts
FROM pipeline_audit
WHERE pipeline_name = %s
AND status = 'success'
ORDER BY completed_at DESC
LIMIT 1
""",
(pipeline_name,),
)
row = cursor.fetchone()
if row is None:
return SLAViolation(
pipeline_name=pipeline_name,
tier=config["tier"],
violation_type="freshness",
description=f"No successful runs found for {pipeline_name}",
severity="critical",
measured_value=float("inf"),
threshold_value=freshness_sla.get("max_lag_minutes",
freshness_sla.get("max_lag_hours", 0) * 60),
)
now = datetime.now(timezone.utc)
reference_ts = row["max_source_ts"] or row["completed_at"]
lag_minutes = (now - reference_ts).total_seconds() / 60
threshold_minutes = freshness_sla.get(
"max_lag_minutes", freshness_sla.get("max_lag_hours", 1) * 60
)
if lag_minutes > threshold_minutes:
severity = "critical" if lag_minutes > threshold_minutes * 2 else "warning"
return SLAViolation(
pipeline_name=pipeline_name,
tier=config["tier"],
violation_type="freshness",
description=(
f"Data is {lag_minutes:.1f} minutes stale "
f"(threshold: {threshold_minutes} minutes). "
f"Last successful run: {row['completed_at'].isoformat()}"
),
severity=severity,
measured_value=lag_minutes,
threshold_value=threshold_minutes,
run_id=row["run_id"],
)
return None
def _check_latency(
self, pipeline_name: str, config: dict, latency_sla: dict
) -> List[SLAViolation]:
"""
Calculate p95 and p99 pipeline durations over the last 24 hours
and compare against thresholds.
"""
cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cursor.execute(
"""
SELECT
EXTRACT(EPOCH FROM (completed_at - started_at)) / 60 AS duration_minutes,
PERCENTILE_CONT(0.95) WITHIN GROUP (
ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at))
) OVER () / 60 AS p95_minutes,
PERCENTILE_CONT(0.99) WITHIN GROUP (
ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at))
) OVER () / 60 AS p99_minutes,
COUNT(*) OVER () AS run_count
FROM pipeline_audit
WHERE pipeline_name = %s
AND status IN ('success', 'partial')
AND completed_at > NOW() - INTERVAL '24 hours'
ORDER BY completed_at DESC
LIMIT 1
""",
(pipeline_name,),
)
row = cursor.fetchone()
violations = []
if row is None or row["run_count"] < 3:
return violations # Not enough data to evaluate
if "p95_minutes" in latency_sla and row["p95_minutes"]:
if row["p95_minutes"] > latency_sla["p95_minutes"]:
violations.append(
SLAViolation(
pipeline_name=pipeline_name,
tier=config["tier"],
violation_type="latency",
description=(
f"p95 latency is {row['p95_minutes']:.1f} min "
f"(threshold: {latency_sla['p95_minutes']} min) "
f"over last 24 hours"
),
severity="warning",
measured_value=row["p95_minutes"],
threshold_value=latency_sla["p95_minutes"],
)
)
return violations
def _check_completeness(
self, pipeline_name: str, config: dict, completeness_sla: dict
) -> Optional[SLAViolation]:
"""
Compare the most recent run's record count against the rolling
average of recent successful runs to detect unexpected drops.
"""
cursor = self.db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
window_days = completeness_sla.get("comparison_window_days", 7)
tolerance_pct = completeness_sla.get("row_count_tolerance_pct", 5.0)
minimum_rows = completeness_sla.get("minimum_rows", 0)
cursor.execute(
"""
WITH recent_runs AS (
SELECT records_written, completed_at,
AVG(records_written) OVER (
ORDER BY completed_at
ROWS BETWEEN %s PRECEDING AND 1 PRECEDING
) AS rolling_avg
FROM pipeline_audit
WHERE pipeline_name = %s
AND status = 'success'
AND completed_at > NOW() - (%s || ' days')::INTERVAL
ORDER BY completed_at DESC
)
SELECT records_written, rolling_avg
FROM recent_runs
LIMIT 1
""",
(window_days, pipeline_name, window_days),
)
row = cursor.fetchone()
if row is None or row["rolling_avg"] is None:
return None # Not enough history
current = row["records_written"] or 0
avg = float(row["rolling_avg"])
if current < minimum_rows:
return SLAViolation(
pipeline_name=pipeline_name,
tier=config["tier"],
violation_type="completeness",
description=(
f"Latest run wrote {current:,} rows, below minimum of "
f"{minimum_rows:,}"
),
severity="critical",
measured_value=current,
threshold_value=minimum_rows,
)
if avg > 0:
deviation_pct = abs(current - avg) / avg * 100
if deviation_pct > tolerance_pct:
severity = "critical" if deviation_pct > tolerance_pct * 2 else "warning"
return SLAViolation(
pipeline_name=pipeline_name,
tier=config["tier"],
violation_type="completeness",
description=(
f"Row count deviation: {deviation_pct:.1f}% from "
f"{window_days}-day average of {avg:,.0f} rows "
f"(current: {current:,}, tolerance: ±{tolerance_pct}%)"
),
severity=severity,
measured_value=deviation_pct,
threshold_value=tolerance_pct,
)
return None
The evaluator produces SLAViolation objects. Now you need to route them to the right people through the right channels:
import requests
from typing import List
class SLAAlertRouter:
def __init__(self, slack_webhook_url: str, pagerduty_routing_key: str = None):
self.slack_webhook_url = slack_webhook_url
self.pagerduty_routing_key = pagerduty_routing_key
def route(self, violations: List[SLAViolation], sla_config: dict):
for violation in violations:
pipeline_config = sla_config["pipelines"].get(violation.pipeline_name, {})
channel = pipeline_config.get("escalation_channel", "#data-alerts")
# Always send to Slack
self._send_slack_alert(violation, channel)
# PagerDuty for critical P0/P1 violations
if (
violation.severity == "critical"
and violation.tier in ("P0", "P1")
and self.pagerduty_routing_key
):
self._send_pagerduty_alert(violation)
def _send_slack_alert(self, violation: SLAViolation, channel: str):
emoji = "🔴" if violation.severity == "critical" else "🟡"
color = "#FF0000" if violation.severity == "critical" else "#FFA500"
payload = {
"channel": channel,
"attachments": [
{
"color": color,
"title": f"{emoji} SLA Violation: {violation.pipeline_name}",
"fields": [
{"title": "Type", "value": violation.violation_type, "short": True},
{"title": "Tier", "value": violation.tier, "short": True},
{"title": "Severity", "value": violation.severity.upper(), "short": True},
{
"title": "Measured",
"value": f"{violation.measured_value:.2f}",
"short": True,
},
{
"title": "Threshold",
"value": f"{violation.threshold_value:.2f}",
"short": True,
},
{"title": "Details", "value": violation.description, "short": False},
],
"footer": f"Detected at {violation.detected_at.strftime('%Y-%m-%d %H:%M UTC')}",
}
],
}
requests.post(self.slack_webhook_url, json=payload, timeout=5)
def _send_pagerduty_alert(self, violation: SLAViolation):
payload = {
"routing_key": self.pagerduty_routing_key,
"event_action": "trigger",
"dedup_key": f"{violation.pipeline_name}_{violation.violation_type}",
"payload": {
"summary": f"SLA Violation [{violation.tier}]: {violation.pipeline_name} - {violation.violation_type}",
"severity": "critical",
"source": "data-pipeline-sla-monitor",
"custom_details": {
"pipeline": violation.pipeline_name,
"violation_type": violation.violation_type,
"description": violation.description,
"measured_value": violation.measured_value,
"threshold": violation.threshold_value,
},
},
}
requests.post(
"https://events.pagerduty.com/v2/enqueue",
json=payload,
timeout=5,
)
Alert fatigue is a real danger. If your SLA monitor fires on every individual run that's slightly slow, your team will start ignoring alerts within a week. Consider adding a "cooldown" mechanism that suppresses duplicate alerts for the same pipeline/violation-type combination for a configurable window (e.g., 30 minutes for P1, 2 hours for P2).
Individual violation alerts are tactical. You also need strategic visibility: are you getting better or worse at meeting your SLAs?
Store SLA violations in their own table (not just alert them), and run weekly compliance reports:
-- SLA compliance summary for the last 30 days
WITH pipeline_runs AS (
SELECT
pipeline_name,
DATE_TRUNC('day', completed_at) AS run_date,
COUNT(*) AS total_runs,
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS successful_runs,
SUM(CASE WHEN status = 'partial' THEN 1 ELSE 0 END) AS partial_runs,
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed_runs,
AVG(
EXTRACT(EPOCH FROM (completed_at - started_at)) / 60
) AS avg_latency_minutes,
PERCENTILE_CONT(0.95) WITHIN GROUP (
ORDER BY EXTRACT(EPOCH FROM (completed_at - started_at)) / 60
) AS p95_latency_minutes
FROM pipeline_audit
WHERE completed_at > NOW() - INTERVAL '30 days'
AND completed_at IS NOT NULL
GROUP BY 1, 2
),
sla_violations AS (
SELECT
pipeline_name,
DATE_TRUNC('day', detected_at) AS violation_date,
COUNT(*) AS violations,
SUM(CASE WHEN severity = 'critical' THEN 1 ELSE 0 END) AS critical_violations
FROM sla_violation_log
WHERE detected_at > NOW() - INTERVAL '30 days'
GROUP BY 1, 2
)
SELECT
r.pipeline_name,
SUM(r.total_runs) AS total_runs_30d,
SUM(r.successful_runs)::FLOAT / NULLIF(SUM(r.total_runs), 0) * 100 AS success_rate_pct,
AVG(r.p95_latency_minutes) AS avg_p95_latency_minutes,
COALESCE(SUM(v.violations), 0) AS total_sla_violations,
COALESCE(SUM(v.critical_violations), 0) AS critical_violations,
(30 - COUNT(DISTINCT v.violation_date))::FLOAT / 30 * 100 AS clean_day_pct
FROM pipeline_runs r
LEFT JOIN sla_violations v ON r.pipeline_name = v.pipeline_name
AND r.run_date = v.violation_date
GROUP BY 1
ORDER BY critical_violations DESC, success_rate_pct ASC;
Run this query weekly and put the results in front of your team. Pipelines with deteriorating compliance become candidates for the next architecture sprint; pipelines with consistently clean records can have their SLA thresholds tightened.
You're a data engineer at a B2B SaaS company. The customer_usage_events pipeline runs every 30 minutes, reads from a Kafka consumer, and writes to a Snowflake table that feeds a real-time customer health dashboard. The sales team uses this dashboard to identify customers at risk of churning.
Your task:
Define an appropriate SLA for this pipeline across all four dimensions. Use the YAML schema from earlier. Assume business hours are 7 AM–8 PM in the customer's local timezone (you can simplify to UTC-8 to UTC+5 range, so effectively 24/7 monitoring is needed).
Add the PipelineAuditContext to a hypothetical pipeline function (you can write the skeleton with placeholder comments for the actual processing logic):
def run_customer_usage_pipeline(kafka_consumer, snowflake_conn, audit_db_conn):
"""
Consume customer usage events from Kafka, transform them,
and write to Snowflake customer_usage_events table.
"""
# Your implementation here — use the PipelineAuditContext
pass
Write the freshness check specifically for this pipeline. The freshness SLA should measure the lag between the Kafka message timestamp (stored in max_source_ts) and wall clock time, not the pipeline completion time. Explain in a comment why this distinction matters for customer health scoring.
Define what a "partial" run means for this specific pipeline — what percentage of record loss would you tolerate before escalating from warning to critical, and why?
Stretch goal: Extend the SLAEvaluator to detect a "stuck pipeline" scenario — where the pipeline appears to be running (status = 'running' in the audit table) but has been running for longer than the p99 latency threshold. This is distinct from a failed pipeline and requires a different response.
This is the single most common SLA instrumentation mistake. If your freshness SLA says "data should be no more than 1 hour old" but you measure pipeline completion time, you'll miss scenarios where the source system itself is delayed. A pipeline that runs perfectly on schedule but receives data that's already 50 minutes old will pass your freshness check while your consumers receive 110-minute-old data.
Fix: Always capture max_source_ts — the highest event timestamp in the batch — and use that as your reference point for freshness calculation.
Setting a p95 latency threshold of 5 minutes because "that sounds fast" when your actual p95 is 12 minutes means you'll be in perpetual violation from day one. Your team starts ignoring alerts, the alerting system loses credibility, and you're back to the Tuesday morning dashboard incident from the introduction.
Fix: Before setting thresholds, run your evaluator in "observation mode" for two weeks — capture what your pipelines actually do, then set thresholds at the 90th percentile of observed performance. Tighten them quarterly.
A completeness check that compares Monday's row count against a 7-day rolling average will always fire on Mondays, because weekday volumes are typically higher than weekend volumes. The check becomes noise.
Fix: Compare against the same day of week. In your rolling average query, filter WHERE EXTRACT(DOW FROM completed_at) = EXTRACT(DOW FROM NOW()).
If your pipeline runs every 5 minutes and your SLA monitor runs every 5 minutes, a 20-minute degradation period will fire four consecutive alerts. Your on-call engineer acknowledges the first, then gets paged three more times while actively investigating.
Fix: Implement deduplication using the dedup_key field in PagerDuty (shown above), or add a "first violation at" timestamp to your violation log and suppress re-alerts until the violation is resolved or a cooldown window expires.
Your SLA monitoring job is a pipeline. It can fail. If it fails silently, you go dark — no violations fire even when pipelines are broken.
Fix: Implement a dead man's switch: your SLA monitor should write a heartbeat record every time it runs successfully. A separate, simple alerting rule fires if no heartbeat arrives within 2x the monitoring interval.
Building effective pipeline SLAs is a discipline that sits at the intersection of engineering and business communication. The technical work — audit tables, context managers, evaluators, alert routers — is genuinely straightforward once you commit to the framework. The harder work is the ongoing negotiation between what your consumers need and what your systems can reliably provide.
Here's what you built in this lesson:
success and partial run statesWhere to go from here:
The logical next step is integrating this system with your data catalog or observability platform. Tools like Monte Carlo, Soda Core, and Great Expectations offer complementary approaches — particularly for data quality checks that go deeper than row counts. Soda Core, notably, has an open-source SLA/freshness checking framework that's worth evaluating against the custom approach built here.
You should also explore how SLA metadata can feed into your data catalog as machine-readable lineage. When a consumer sees a table in your catalog, they should be able to click through to its SLA definition, its current compliance status, and its recent violation history. That kind of transparency is what turns a data engineering team from a black box into a trustworthy infrastructure provider.
Finally, revisit your SLAs quarterly. Business priorities shift, data volumes grow, and pipeline architectures evolve. An SLA that was perfectly calibrated six months ago may be either too loose (masking real degradation) or too tight (generating chronic noise) today. Treat your SLA thresholds the way a good engineer treats any configuration: version-controlled, reviewed, and deliberately updated based on evidence.
Learning Path: Data Pipeline Fundamentals