Data observability: a practical Spark & Data Eng guide

Introduction

We've spent fifty-nine days building data pipelines, optimizing Spark jobs, and managing infrastructure. Now we need to treat those pipelines the way we treat production software: with monitoring, alerting, and accountability. Data observability: a practical Spark & Data Eng guide is about seeing into your data ecosystem in real time—knowing when freshness drops, when volume anomalies occur, when schemas break, and who owns the problem.

Most teams discover data issues through angry Slack messages from downstream consumers. That's reactive. Observability moves us to proactive detection, faster incident response, and systemic learning. Unlike traditional infrastructure monitoring that watches CPU and memory, data observability watches the data itself.

Why data observability matters

Data pipelines fail silently. A Spark job completes successfully and reports no errors, yet the output table has zero rows. Your ML model scores silently on stale features. A downstream dashboard shows numbers that contradict yesterday's report. Without observability, teams waste hours on triage and lose trust in data.

Treating data like production software means accepting that data incidents happen—and preparing for them with the same discipline we use for application outages. That means instrumentation, monitoring, alerting, and postmortems.

Freshness monitors: knowing your data currency

Freshness is the time between when data should be available and when it actually is. A daily ETL that should land by 8 AM but consistently arrives at 9:30 AM is stale. A real-time streaming pipeline dropping to hour-old data needs attention.

Set freshness expectations based on downstream SLAs, not on how fast your infrastructure can run. If a dashboard is consumed at 9 AM and depends on yesterday's data, that data must land by 8:55 AM. Build that into your contract.

A practical freshness monitor tracks the maximum timestamp of data in a table and compares it to the current time:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, max, current_timestamp
from datetime import datetime, timedelta

spark = SparkSession.builder.appName("FreshnessMonitor").getOrCreate()

# Check table freshness
table_name = "events"
max_event_time = spark.read.table(table_name) \
    .select(max(col("event_timestamp")).alias("max_ts")) \
    .collect()[0]["max_ts"]

current_time = datetime.utcnow()
staleness_minutes = (current_time - max_event_time).total_seconds() / 60

freshness_threshold = 30  # minutes
is_fresh = staleness_minutes < freshness_threshold

# Log to observability backend
print(f"Table: {table_name}")
print(f"Max event time: {max_event_time}")
print(f"Staleness: {staleness_minutes} minutes")
print(f"Status: {'FRESH' if is_fresh else 'STALE'}")

In production, this runs as a scheduled job (every 5–10 minutes), and the result feeds into your alerting system. Different tables have different thresholds—a user profile table might tolerate 24-hour staleness; a transaction table cannot.

Volume monitors: detecting silent failures

A Spark job completes without error. The output table exists. But it has 50% fewer rows than yesterday, and nobody noticed for six hours. Volume monitoring detects these silent failures.

Volume checks go beyond "did rows appear." They look for anomalies: is today's volume within expected variance of recent history? Is it consistent with what upstream produced?

from pyspark.sql.functions import col, count, stddev_pop, avg

# Volume check for a table
table_df = spark.read.table("orders")
current_row_count = table_df.count()

# Compare against historical baseline
baseline_stats = spark.sql("""
    SELECT
        DATE(order_date) as date,
        COUNT(*) as row_count
    FROM orders
    WHERE DATE(order_date) >= CURRENT_DATE - INTERVAL 30 DAY
        AND DATE(order_date) < CURRENT_DATE
    GROUP BY DATE(order_date)
""")

baseline_values = baseline_stats.select("row_count").rdd.flatMap(lambda x: x).collect()
baseline_mean = sum(baseline_values) / len(baseline_values)
baseline_stddev = (sum((x - baseline_mean) ** 2 for x in baseline_values) / len(baseline_values)) ** 0.5

# Z-score check
z_score = (current_row_count - baseline_mean) / (baseline_stddev + 1)
is_anomalous = abs(z_score) > 2  # 2 standard deviations

print(f"Current row count: {current_row_count}")
print(f"Baseline mean: {baseline_mean:.0f}")
print(f"Z-score: {z_score:.2f}")
print(f"Anomalous: {is_anomalous}")

This approach uses standard deviation to account for natural variance. If your pipeline processes 100K rows daily but hits 200K on Mondays and 50K on Sundays, a fixed threshold fails. A statistical approach adapts to real patterns.

Schema monitors: catching breaking changes early

A column type changes from string to int. A required field is suddenly null. A column gets dropped. These schema changes break downstream jobs silently because Spark is forgiving about mismatched types.

Track schema evolution explicitly. Capture the schema of each dataset at ingestion time, and alert if it drifts from what you declared:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Expected schema for user events
expected_schema = StructType([
    StructField("user_id", IntegerType(), False),
    StructField("event_type", StringType(), False),
    StructField("event_time", StringType(), False),
    StructField("properties", StringType(), True),
])

# Read incoming data
incoming_df = spark.read.json("s3://raw-events/latest/")
actual_schema = incoming_df.schema

# Compare schemas
expected_fields = {f.name: (f.dataType, f.nullable) for f in expected_schema.fields}
actual_fields = {f.name: (f.dataType, f.nullable) for f in actual_schema.fields}

issues = []

# Check for missing or changed fields
for field_name, (expected_type, expected_nullable) in expected_fields.items():
    if field_name not in actual_fields:
        issues.append(f"Missing field: {field_name}")
    else:
        actual_type, actual_nullable = actual_fields[field_name]
        if str(expected_type) != str(actual_type):
            issues.append(f"Type mismatch for {field_name}: expected {expected_type}, got {actual_type}")
        if expected_nullable != actual_nullable:
            issues.append(f"Nullability mismatch for {field_name}")

# Check for new unexpected fields
for field_name in actual_fields:
    if field_name not in expected_fields:
        issues.append(f"Unexpected field: {field_name}")

if issues:
    print("SCHEMA DRIFT DETECTED:")
    for issue in issues:
        print(f"  - {issue}")
else:
    print("Schema matches expectations")

Store the expected schema in version control or a metadata store. Make schema changes go through a review process before they hit production. This is the data equivalent of API contract tests.

Lineage for impact analysis

When a dataset fails, knowing which downstream consumers depend on it determines urgency and scope. Lineage answers the question: if this table breaks, what else breaks?

Implement lineage tracking at the job level. Every Spark job should record its inputs and outputs:

class DataLineageTracker:
    def __init__(self, job_name, observability_backend):
        self.job_name = job_name
        self.backend = observability_backend
        self.inputs = []
        self.outputs = []
    
    def add_input(self, table_name, database="default"):
        self.inputs.append(f"{database}.{table_name}")
    
    def add_output(self, table_name, database="default"):
        self.outputs.append(f"{database}.{table_name}")
    
    def record(self):
        """Send lineage to backend (DataHub, Atlas, or custom)"""
        lineage_event = {
            "job": self.job_name,
            "inputs": self.inputs,
            "outputs": self.outputs,
            "timestamp": datetime.utcnow().isoformat(),
        }
        self.backend.send(lineage_event)

# Usage in your Spark pipeline
tracker = DataLineageTracker("orders_aggregation", observability_backend)
tracker.add_input("raw_orders")
tracker.add_input("order_items")
tracker.add_output("daily_orders")
tracker.record()

# Your Spark job here
orders_df = spark.read.table("raw_orders")
items_df = spark.read.table("order_items")
result = orders_df.join(items_df, "order_id").groupBy("date").count()
result.write.mode("overwrite").saveAsTable("daily_orders")

With complete lineage, when raw_orders fails to load, you can immediately identify all affected jobs and notify their owners. Lineage also powers impact analysis—before a breaking schema change, you see exactly which downstream teams will be affected.

Alerts routed to owners

Monitoring without alerting is just logging. Alerts are worthless if they land in a channel nobody reads or if they reach the wrong team.

Design your alerting with ownership in mind. Each dataset should have a clear owner or team. Alerts should route based on severity and ownership:

• Critical freshness violations (more than 2 hours stale): immediate Slack to on-call and escalation if not resolved in 15 minutes
• Volume anomalies (±50% of baseline): Slack notification to data team for triage
• Schema drift: Slack to owning team plus data engineering; blocks pipeline execution
• Quality rule failures (nulls in required columns, duplicates): logged and reviewed in daily standup

Implement alert fatigue controls. Too many alerts and teams stop reading them. Use progressive alerting: for marginal issues, log them; for clear problems, alert; for critical failures, escalate. Tune thresholds based on false positive rate—if an alert fires 50 times a week and is wrong half the time, it's noise.

Incidents and postmortems for data

When a data incident happens—stale tables, wrong totals, missing records—treat it like a production incident. Create a ticket, investigate, and conduct a postmortem.

Postmortems are not blame exercises. They're learning opportunities. Did the alert fire but get ignored? Did the owner not have runbooks? Did the monitoring check itself have a bug?

A typical data incident postmortem covers:

• Timeline: when data went wrong, when it was detected, when it was fixed
• Impact: how many records affected, which consumers impacted, business cost
• Root cause: what actually broke—upstream dependency failed? Schema changed? Spark job hit timeout?
• Detection gap: why didn't we catch it sooner? Was monitoring insufficient?
• Action items: concrete changes to prevent recurrence (add monitor, fix job, update runbook)

Track action items in a backlog. Make fixing data observability gaps a priority—a monitor that would have caught an incident in five minutes is worth building now.

Common mistakes

• Monitoring only success metrics. Spark jobs complete successfully but with wrong data. Monitor volume, freshness, and schema independently of job exit codes.
• Generic thresholds across all tables. A 24-hour delay for daily summaries is acceptable; for real-time event streams, it's a failure. Tune thresholds per dataset based on SLA.
• Alerts with no context. An alert saying "high cardinality on user_id" without linking to runbooks, owners, or recent changes wastes time. Include context in every alert.
• Treating lineage as optional. You can't debug a cascade of failures without knowing dependencies. Build lineage tracking into every pipeline from the start.
• Skipping postmortems for "minor" data issues. Small issues are practice for big ones. Every incident is a chance to improve observability.
• Centralizing all ownership. If the data platform team owns alerting and runbooks, they become a bottleneck. Give data producers the tools and context to own their own pipelines.

Key takeaways

• Treat data pipelines as production systems: monitor freshness, volume, and schema continuously
• Use statistical approaches (z-scores, baselines) for anomaly detection rather than fixed thresholds
• Track data lineage at the job level so you know impact of failures before they cascade
• Route alerts based on ownership and severity; tune to avoid alert fatigue
• Conduct postmortems on data incidents to close detection and prevention gaps
• Make observability the responsibility of data producers, not just platform teams