De Root Cause Analysis

DE Root Cause Analysis

When to Use

Load this skill when the user needs to:

• Triage a failing or stuck data pipeline (Airflow DAG, Spark job, dbt run)
• Investigate a data quality incident — wrong numbers, missing rows, duplicates, nulls
• Analyze an SLA breach — pipeline ran but finished too late
• Respond to a production incident — unexpected data anomaly reported by stakeholders
• Trace blast radius — which downstream tables/dashboards/jobs are affected by a bad upstream table
• Write an RCA document or post-mortem after resolving an incident
• Build investigation checklists and runbooks for on-call engineers

1. Failure Taxonomy

Correctly classifying the failure type determines the fastest investigation path. Misclassifying (e.g., treating a logic bug as an infrastructure problem) wastes hours.

1.1 Infrastructure Failures

1.2 Data Failures

1.3 Logic Failures

1.4 Dependency Failures

1.5 Configuration Failures

1.6 Concurrency Failures

2. Investigation Framework — 5-Step RCA Process

Step 1: Timeline Reconstruction

The timeline is the foundation of all RCA. Establish three timestamps first:

1. Event time — when did the bad data/failure actually occur in the data?
2. Detection time — when did a monitor, alert, or human notice?
3. Impact time — when did downstream consumers first see the problem?

-- Airflow: find first failure for a DAG in the last 7 days
SELECT
    dag_id,
    run_id,
    execution_date,
    state,
    start_date,
    end_date,
    (end_date - start_date) AS duration
FROM dag_run
WHERE dag_id = 'orders_etl'
  AND execution_date >= NOW() - INTERVAL '7 days'
ORDER BY execution_date DESC;

-- Find which task instance failed first in a run
SELECT
    task_id,
    state,
    start_date,
    end_date,
    try_number,
    hostname
FROM task_instance
WHERE dag_id   = 'orders_etl'
  AND run_id   = 'scheduled__2024-06-01T00:00:00+00:00'
ORDER BY start_date;

Step 2: Blast Radius Assessment

Before digging into root cause, know what is broken. This prevents partial fixes and missed downstream alerts.

# Quick blast radius checklist
BLAST_RADIUS_QUESTIONS = [
    "Which tables are directly written by the failing job?",
    "Which dbt models SELECT from those tables?",
    "Which dashboards/BI tools query those models?",
    "Which downstream DAGs depend on this DAG (ExternalTaskSensor)?",
    "Which SLAs are now at risk given current time?",
    "Are any customer-facing APIs/reports backed by this data?",
]

# dbt: find all models that depend on a broken source model
dbt ls --select "source:raw.orders+" --output path
# + suffix means "this node and all descendants"

# dbt: find direct and indirect dependents of a specific model
dbt ls --select "+stg_orders+" --output name

Step 3: Evidence Collection

Collect evidence before forming hypotheses — evidence is objective, hypotheses are not.

Checklist:

• Task/job logs (full log, not just last N lines)
• Row counts before and after the failing step
• Sample of affected rows (5-20 rows is usually enough)
• Metric graphs for the failure window (Grafana, CloudWatch, Datadog)
• Schema of source and target tables at failure time
• Git history of pipeline code — any deploy in the last 24h?
• Infrastructure change log — cluster resize, config push, dependency upgrade?
• Upstream system status page or on-call contact

Step 4: Hypothesis Generation and Elimination

Form hypotheses from the failure taxonomy (Section 1), then eliminate them systematically with evidence.

Hypothesis template:
  "The failure was caused by [X] because [evidence Y] supports it.
   It was NOT caused by [A] because [evidence B] rules it out."

Good elimination requires a binary question: "If hypothesis X is true, what observable artifact would exist?" If that artifact is absent, eliminate X.

Step 5: Root Cause Statement and Contributing Factors

A root cause statement is a single sentence using the because chain:

"The orders dashboard showed zero revenue on 2024-06-01 because the fact_orders dbt model produced zero rows because the incremental filter used updated_at > '{{ ds }}' with ds in UTC while source timestamps are in US/Eastern, excluding all rows written after 19:00 UTC the previous day."

Contributing factors are conditions that made the root cause possible or worse:

• No data volume alert on fact_orders to catch the zero-row scenario
• Timezone assumption was undocumented in the model's description
• The issue was not caught in staging because the staging run uses a small sample

3. Airflow Failure Diagnosis

3.1 Task Log Location and Key Patterns

# Default log path on Airflow worker
/opt/airflow/logs/<dag_id>/<task_id>/<execution_date>/<try_number>.log

# Kubernetes executor — fetch logs from pod
kubectl logs <pod-name> -n airflow -c base

# Grep for common failure signals in a task log
grep -E "(ERROR|CRITICAL|Traceback|Exception|killed|OOM|timeout)" task.log

# Find the actual exception line (last Traceback in log)
grep -n "Traceback" task.log | tail -1
# Then read from that line number
tail -n +<LINE> task.log | head -60

Structured log patterns to search (JSON logs):

# jq: extract ERROR entries with timestamps
cat task.log | jq -r 'select(.level=="ERROR") | "\(.timestamp) \(.event // .message)"'

# jq: find the first error occurrence
cat task.log | jq -r 'select(.level=="ERROR") | .timestamp' | head -1

3.2 Task Instance State Machine

-- Find tasks stuck in running state for more than 2 hours (likely zombie)
SELECT dag_id, task_id, run_id, start_date, hostname
FROM task_instance
WHERE state = 'running'
  AND start_date < NOW() - INTERVAL '2 hours';

-- Failure rate per task over last 30 days
SELECT
    dag_id,
    task_id,
    COUNT(*)                                           AS total_runs,
    SUM(CASE WHEN state = 'failed' THEN 1 ELSE 0 END) AS failures,
    ROUND(100.0 * SUM(CASE WHEN state = 'failed' THEN 1 ELSE 0 END) / COUNT(*), 2) AS failure_pct,
    MAX(CASE WHEN state = 'failed' THEN start_date END) AS last_failure
FROM task_instance
WHERE start_date >= NOW() - INTERVAL '30 days'
GROUP BY dag_id, task_id
HAVING SUM(CASE WHEN state = 'failed' THEN 1 ELSE 0 END) > 0
ORDER BY failure_pct DESC;

3.3 DagBag Import Errors

DagBag errors prevent the DAG from appearing in the UI at all — the scheduler silently skips broken DAGs.

# Reproduce a DagBag import error locally
python -c "
from airflow.models import DagBag
bag = DagBag(dag_folder='/opt/airflow/dags', include_examples=False)
for dag_id, err in bag.import_errors.items():
    print(f'--- {dag_id} ---')
    print(err)
"

# Or using the CLI
airflow dags list-import-errors

# Test a single DAG file directly
python /opt/airflow/dags/orders_etl.py

Common DagBag import errors:

• ModuleNotFoundError — missing Python package in the Airflow environment
• SyntaxError — Python syntax error in the DAG file
• NameError / AttributeError — wrong import path or API version mismatch (Airflow 2 vs 3)
• Circular dag_id — two files define the same dag_id

3.4 Distinguishing Sensor Failures

# Sensor timeout (total_timeout exceeded) — log signature:
# airflow.exceptions.AirflowSensorTimeout: Snap. Time is OUT.
# Fix: increase total_timeout or investigate why condition never became true

# Poke failure (poke() returned False indefinitely) — the sensor keeps retrying
# until timeout. Check: is the file/API/task actually completing?

# Operator failure (exception inside execute()) — stack trace visible in logs
# This is a code/config bug, not a sensor timing issue

# ExternalTaskSensor stuck: upstream DAG run doesn't exist for the execution_date
# Check: does the parent DAG use the same schedule and execution_date?
SELECT dag_id, execution_date, state
FROM dag_run
WHERE dag_id = 'upstream_dag'
  AND execution_date = '2024-06-01 00:00:00';

3.5 XCom Size Limit Causing Silent Failures

# Airflow stores XComs in the metadata DB (default: 48KB limit in MySQL, 1GB in PostgreSQL)
# Large XComs silently truncate or raise OperationalError

# Detect oversized XCom push
from airflow.sdk import task
import sys

@task
def bad_xcom_task() -> list:
    data = list(range(1_000_000))   # ~8MB — will fail or truncate
    print(f"XCom size estimate: {sys.getsizeof(data)} bytes")
    return data  # DO NOT push large objects via XCom

# Fix: write to S3/GCS, push only the path
@task
def good_xcom_task() -> str:
    data = list(range(1_000_000))
    path = "s3://my-bucket/tmp/task_output.parquet"
    _write_to_s3(data, path)
    return path   # push only the reference

3.6 Worker OOM vs Task Timeout

Worker OOM signature in logs:
  "Killed" (from Linux OOM killer — no Python traceback)
  "Worker exited prematurely: signal 9 (SIGKILL)"
  "Container killed by YARN due to exceeding memory"

Task timeout signature in logs:
  "airflow.exceptions.AirflowTaskTimeout: Timeout"
  "DagRunTimeoutError" or "execution_timeout reached"
  Clean Python traceback visible

OOM fix: reduce memory footprint (chunked reads, Spark executor memory tuning)
Timeout fix: increase execution_timeout or optimize the slow step

4. Spark/PySpark Failure Diagnosis

4.1 Reading Spark UI

Navigate to: http://<driver-host>:4040 (active job) or Spark History Server for completed jobs.

Key pages:

• Jobs — overall job status; failed jobs shown in red
• Stages — click a failed stage to see the failure reason
• Failed stage > Tasks tab — sort by Error column; first failed task has the root cause
• Executors tab — check GC Time % (>10% means memory pressure), spill to disk, failed tasks per executor
• SQL tab — query plan with row counts per node; large row count jumps indicate fan-out from bad joins

4.2 OOM: Executor vs Driver

Executor OOM:
  java.lang.OutOfMemoryError: Java heap space       (in executor log)
  Container killed by YARN for exceeding memory limits
  Error: ExecutorLostFailure (executor 3 exited caused by: Executor OOM)
  Fix:
    spark.executor.memory          = 4g -> 8g
    spark.executor.memoryOverhead  = 512m -> 1g
    spark.sql.shuffle.partitions   = 200 -> 800  (smaller partitions)
    repartition(n) before heavy shuffle

Driver OOM:
  java.lang.OutOfMemoryError in driver logs
  "Driver exited with exit code 1"
  Caused by: collect(), toPandas(), broadcast of large DF
  Fix:
    spark.driver.memory            = 2g -> 8g
    Replace collect() with write to storage
    Replace broadcast join with sort-merge join if table > 100MB

4.3 Data Skew Detection

from pyspark.sql import functions as F

def detect_skew(df, join_key: str, top_n: int = 20):
    """Print key distribution to find skew."""
    dist = (
        df.groupBy(join_key)
          .agg(F.count("*").alias("row_count"))
          .orderBy(F.desc("row_count"))
    )
    print(f"Top {top_n} values for '{join_key}':")
    dist.show(top_n, truncate=False)

    stats = df.groupBy(join_key).count().agg(
        F.min("count").alias("min"),
        F.avg("count").alias("avg"),
        F.max("count").alias("max"),
        F.stddev("count").alias("stddev"),
    )
    stats.show()
    # Skew indicator: max / avg > 10x

# Fix skew with salting
import pyspark.sql.functions as F
from pyspark.sql import DataFrame

SALT_BUCKETS = 50

def salt_join(large_df: DataFrame, small_df: DataFrame, key: str) -> DataFrame:
    """Salt-based skew join: explode small_df, add random salt to large_df."""
    large_salted = large_df.withColumn(
        "salt_key",
        F.concat(F.col(key).cast("string"), F.lit("_"),
                 (F.rand() * SALT_BUCKETS).cast("int").cast("string"))
    )
    small_exploded = small_df.withColumn(
        "salt_key",
        F.explode(F.array([
            F.concat(F.col(key).cast("string"), F.lit(f"_{i}"))
            for i in range(SALT_BUCKETS)
        ]))
    )
    return large_salted.join(small_exploded, "salt_key").drop("salt_key")

4.4 Serialization Errors

Error signature:
  org.apache.spark.SparkException: Task not serializable
  Caused by: java.io.NotSerializableException: com.example.MyClass

Root causes:
  1. Lambda captures a non-serializable object (DB connection, file handle)
  2. UDF references an instance variable of a non-serializable class
  3. Broadcast variable holds a non-serializable object

Fix:

# BAD: captures self (often non-serializable)
class Processor:
    def __init__(self, multiplier):
        self.multiplier = multiplier

    def process(self, df):
        # self is captured — if Processor is not serializable, this fails
        return df.rdd.map(lambda row: row.value * self.multiplier)

# GOOD: extract the primitive value before passing to RDD/UDF
class Processor:
    def __init__(self, multiplier):
        self.multiplier = multiplier

    def process(self, df):
        m = self.multiplier   # extract serializable primitive
        return df.rdd.map(lambda row: row.value * m)

4.5 Shuffle Failures (FetchFailed)

Error signature:
  org.apache.spark.shuffle.FetchFailed: Failed to connect to <host>/<ip>:port

Root causes:
  - Executor that held shuffle data was evicted/OOM killed before downstream stage fetched it
  - Network timeout during large shuffle
  - Disk full on shuffle directory

Diagnosis:
  1. Check Executors tab: was the executor with shuffle data lost?
  2. Check worker disk usage: df -h /tmp on all workers
  3. Check spark.reducer.maxReqsInFlight and spark.shuffle.io.maxRetries

Fixes:
  spark.shuffle.io.maxRetries          = 3  -> 10
  spark.shuffle.io.retryWait           = 5s -> 30s
  spark.sql.shuffle.partitions         increase to reduce per-partition size
  spark.shuffle.service.enabled        = true  (external shuffle service)
  Consider Delta/Iceberg intermediate writes instead of in-memory shuffle

4.6 Schema Mismatch on Read

# AnalysisException: cannot resolve column 'order_id' given input columns
# Causes:
#   - Column renamed in source
#   - Reading Parquet written with different schema evolution settings
#   - Hive metastore schema cached but underlying files differ

# Diagnose schema at read time
df = spark.read.parquet("s3://bucket/orders/")
df.printSchema()

# Compare with expected schema
from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType

EXPECTED_SCHEMA = StructType([
    StructField("order_id",    StringType(),    nullable=False),
    StructField("customer_id", LongType(),      nullable=False),
    StructField("amount",      DoubleType(),    nullable=True),
    StructField("created_at",  TimestampType(), nullable=False),
])

actual_fields   = {f.name: f.dataType for f in df.schema.fields}
expected_fields = {f.name: f.dataType for f in EXPECTED_SCHEMA.fields}

missing  = set(expected_fields) - set(actual_fields)
extra    = set(actual_fields) - set(expected_fields)
mistyped = {k for k in actual_fields if k in expected_fields
            and type(actual_fields[k]) != type(expected_fields[k])}

print(f"Missing columns: {missing}")
print(f"Extra columns:   {extra}")
print(f"Type mismatches: {mistyped}")

5. dbt Failure Diagnosis

5.1 Failure Types

# Reproduce a compilation error locally
dbt compile --select orders_mart
cat target/compiled/my_project/models/orders_mart.sql

# Run a single model with full output
dbt run --select orders_mart --full-refresh 2>&1 | tee /tmp/dbt_run.log

# Run only failing tests for a model
dbt test --select orders_mart 2>&1 | grep "FAIL"

5.2 Reading run_results.json

import json
from pathlib import Path

def parse_dbt_run_results(target_dir: str = "target") -> list[dict]:
    path = Path(target_dir) / "run_results.json"
    data = json.loads(path.read_text())

    failures = []
    for result in data["results"]:
        if result["status"] not in ("success", "pass", "warn"):
            failures.append({
                "unique_id": result["unique_id"],
                "status":    result["status"],
                "message":   result.get("message", ""),
                "timing":    result.get("timing", []),
            })
    return failures

# Usage in CI
failures = parse_dbt_run_results()
for f in failures:
    print(f"FAILED: {f['unique_id']}")
    print(f"  Status:  {f['status']}")
    print(f"  Message: {f['message'][:500]}")

5.3 Incremental Model Issues

-- Common bug: is_incremental() filter uses wrong watermark column
-- If source has both created_at and updated_at, use updated_at for incrementals
-- BAD: misses rows that were updated after creation
{% if is_incremental() %}
WHERE created_at > (SELECT MAX(created_at) FROM {{ this }})
{% endif %}

-- GOOD: catches updates
{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

-- Diagnose unique key violation in incremental merge strategy
-- Run this against the staging data before the dbt run
SELECT order_id, COUNT(*) AS cnt
FROM {{ ref('stg_orders') }}
GROUP BY order_id
HAVING COUNT(*) > 1
ORDER BY cnt DESC
LIMIT 20;

5.4 dbt Test Failure Triage

# Run tests and show the failing SQL
dbt test --select orders_mart --store-failures

# dbt stores failures in the database — query them
# Table name: <schema>_dbt_test__audit.<test_name>

# Find which rows violated not_null test
SELECT *
FROM dbt_test__audit.not_null_orders_mart_order_id
LIMIT 20;

# Find which rows violated unique test
SELECT *
FROM dbt_test__audit.unique_orders_mart_order_id
LIMIT 20;

# Programmatic test result check from CI
import subprocess, json, sys

result = subprocess.run(
    ["dbt", "test", "--select", "orders_mart", "--output", "json"],
    capture_output=True, text=True
)
for line in result.stdout.splitlines():
    try:
        ev = json.loads(line)
        if ev.get("info", {}).get("level") == "error":
            print(ev["info"]["msg"])
    except json.JSONDecodeError:
        pass
if result.returncode != 0:
    sys.exit(1)

5.5 Snapshot Failures

-- Snapshot fails if the configured unique_key is not actually unique in the source
-- Diagnose before running snapshot:
SELECT order_id, COUNT(*) AS cnt
FROM {{ source('raw', 'orders') }}
GROUP BY order_id
HAVING COUNT(*) > 1;

-- Snapshot fails if updated_at column contains NULLs (for timestamp strategy)
SELECT COUNT(*) AS null_updated_at
FROM {{ source('raw', 'orders') }}
WHERE updated_at IS NULL;

6. Data Quality Anomaly Investigation

6.1 Volume Anomaly — Today vs 7-Day Rolling Average

-- PostgreSQL / Trino
WITH daily_counts AS (
    SELECT
        event_date,
        COUNT(*) AS row_count
    FROM orders
    WHERE event_date >= CURRENT_DATE - INTERVAL '8 days'
    GROUP BY event_date
),
stats AS (
    SELECT
        event_date,
        row_count,
        AVG(row_count) OVER (
            ORDER BY event_date
            ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
        ) AS rolling_7d_avg,
        STDDEV(row_count) OVER (
            ORDER BY event_date
            ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
        ) AS rolling_7d_stddev
    FROM daily_counts
)
SELECT
    event_date,
    row_count,
    ROUND(rolling_7d_avg, 0)    AS avg_7d,
    ROUND(rolling_7d_stddev, 0) AS stddev_7d,
    ROUND(100.0 * (row_count - rolling_7d_avg) / NULLIF(rolling_7d_avg, 0), 1) AS pct_vs_avg,
    CASE
        WHEN ABS(row_count - rolling_7d_avg) > 3 * rolling_7d_stddev THEN 'ANOMALY'
        WHEN ABS(row_count - rolling_7d_avg) > 2 * rolling_7d_stddev THEN 'WARNING'
        ELSE 'OK'
    END AS status
FROM stats
ORDER BY event_date DESC;

6.2 Freshness Anomaly — Late vs Missing Data

-- Distinguish late-arriving data from completely missing data
-- Trino / PostgreSQL

-- Check 1: is there ANY data for today (missing)?
SELECT
    COUNT(*) AS today_rows,
    MIN(created_at) AS first_event,
    MAX(created_at) AS last_event
FROM orders
WHERE DATE(created_at) = CURRENT_DATE;

-- Check 2: what is the maximum lag between event_time and load_time?
SELECT
    PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY load_lag_minutes) AS p50_lag_min,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY load_lag_minutes) AS p95_lag_min,
    PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY load_lag_minutes) AS p99_lag_min,
    MAX(load_lag_minutes) AS max_lag_min
FROM (
    SELECT
        EXTRACT(EPOCH FROM (loaded_at - event_time)) / 60.0 AS load_lag_minutes
    FROM orders
    WHERE DATE(event_time) = CURRENT_DATE
) t;

6.3 Distribution Shift

-- Compare min/max/avg/stddev between two time periods
-- Spark SQL / Trino / PostgreSQL
SELECT
    'today'     AS period,
    MIN(amount) AS min_val,
    MAX(amount) AS max_val,
    AVG(amount) AS avg_val,
    STDDEV(amount) AS stddev_val,
    COUNT(*) AS row_count
FROM orders WHERE event_date = CURRENT_DATE
UNION ALL
SELECT
    'yesterday' AS period,
    MIN(amount),
    MAX(amount),
    AVG(amount),
    STDDEV(amount),
    COUNT(*)
FROM orders WHERE event_date = CURRENT_DATE - 1;

6.4 Null Explosion Detection

# PySpark: per-column null rate comparison across two DataFrames
from pyspark.sql import functions as F, DataFrame

def null_rate_comparison(df_today: DataFrame, df_yesterday: DataFrame) -> None:
    """Print columns where null rate changed by more than 5 percentage points."""
    cols = df_today.columns

    def null_rates(df: DataFrame) -> dict:
        total = df.count()
        exprs = [
            F.round(F.sum(F.col(c).isNull().cast("int")) / total * 100, 2).alias(c)
            for c in cols
        ]
        return df.agg(*exprs).collect()[0].asDict()

    today_rates     = null_rates(df_today)
    yesterday_rates = null_rates(df_yesterday)

    print(f"{'Column':<40} {'Yesterday%':>12} {'Today%':>10} {'Delta':>8}")
    print("-" * 74)
    for col in sorted(cols):
        prev = yesterday_rates.get(col, 0) or 0
        curr = today_rates.get(col, 0) or 0
        delta = curr - prev
        flag = " <-- ALERT" if abs(delta) > 5 else ""
        print(f"{col:<40} {prev:>12.2f} {curr:>10.2f} {delta:>+8.2f}{flag}")

-- PostgreSQL: null rate per column for a specific table partition
SELECT
    attname AS column_name,
    null_frac * 100 AS null_pct
FROM pg_stats
WHERE tablename = 'orders'
  AND schemaname = 'public'
ORDER BY null_frac DESC;

6.5 Duplicate Explosion with Source Tracing

-- Find duplicates and trace them to source batch/file
SELECT
    order_id,
    COUNT(*)                            AS dup_count,
    MIN(loaded_at)                      AS first_seen,
    MAX(loaded_at)                      AS last_seen,
    COUNT(DISTINCT source_batch_id)     AS distinct_batches,
    ARRAY_AGG(DISTINCT source_batch_id) AS batch_ids
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1
ORDER BY dup_count DESC
LIMIT 50;

-- Trino / Spark SQL: check if duplicates are exact row duplicates or key-only
WITH dupes AS (
    SELECT order_id
    FROM orders
    GROUP BY order_id
    HAVING COUNT(*) > 1
)
SELECT o.*
FROM orders o
JOIN dupes d ON o.order_id = d.order_id
ORDER BY o.order_id, o.loaded_at;

7. Lineage-Based Impact Analysis

7.1 dbt — Find All Downstream Models

# All nodes downstream of a model (+ means "and all descendants")
dbt ls --select "orders_mart+" --output name

# All nodes upstream and downstream (full lineage path)
dbt ls --select "+orders_mart+" --output name

# Find which models use a specific source
dbt ls --select "source:raw.orders+" --output path

# Generate a lineage graph to a file (requires dbt-artifacts or dbt-docs)
dbt docs generate && dbt docs serve

7.2 OpenLineage / Marquez API

import requests

MARQUEZ_URL = "http://marquez:5000"

def get_downstream_datasets(namespace: str, job_name: str) -> list[dict]:
    """Return datasets produced by this job — these are the immediate blast radius."""
    resp = requests.get(
        f"{MARQUEZ_URL}/api/v1/namespaces/{namespace}/jobs/{job_name}",
        timeout=10,
    )
    resp.raise_for_status()
    job = resp.json()
    return job.get("outputs", [])  # list of {namespace, name} dicts


def get_dataset_lineage(namespace: str, dataset_name: str, depth: int = 3) -> dict:
    """Get complete upstream + downstream lineage graph for a dataset."""
    resp = requests.get(
        f"{MARQUEZ_URL}/api/v1/lineage",
        params={"nodeId": f"dataset:{namespace}:{dataset_name}", "depth": depth},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()  # {graph: [{id, type, data, inEdges, outEdges}]}


# Example: find all jobs that READ from a broken table
def find_consumers(namespace: str, broken_table: str) -> list[str]:
    lineage = get_dataset_lineage(namespace, broken_table, depth=5)
    consumers = []
    for node in lineage.get("graph", []):
        if node["type"] == "JOB":
            for edge in node.get("inEdges", []):
                if broken_table in edge["origin"]:
                    consumers.append(node["id"])
    return consumers

7.3 DataHub GraphQL — Downstream Lineage

import requests

DATAHUB_URL = "http://datahub-gms:8080"

LINEAGE_QUERY = """
query GetDownstreamLineage($urn: String!) {
  dataset(urn: $urn) {
    downstream: lineage(input: {direction: DOWNSTREAM, count: 100}) {
      relationships {
        entity {
          urn
          ... on Dataset {
            name
            platform { name }
            schemaMetadata { fields { fieldPath } }
          }
        }
      }
    }
  }
}
"""

def get_downstream_lineage(dataset_urn: str) -> list[str]:
    resp = requests.post(
        f"{DATAHUB_URL}/api/graphql",
        json={"query": LINEAGE_QUERY, "variables": {"urn": dataset_urn}},
        headers={"Authorization": f"Bearer {DATAHUB_TOKEN}"},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    relationships = (
        data.get("data", {})
            .get("dataset", {})
            .get("downstream", {})
            .get("relationships", [])
    )
    return [r["entity"]["urn"] for r in relationships]

7.4 Quick Manual Impact List

When automated lineage tools are unavailable:

# Grep dbt project SQL files for references to a broken table
grep -r "ref('orders_mart')" dbt_project/models/ --include="*.sql" -l
grep -r "source('raw', 'orders')" dbt_project/models/ --include="*.sql" -l

# Search Airflow DAGs for references to a table name
grep -r "orders_mart" airflow/dags/ --include="*.py" -l

# Search SQL files for a table name
find . -name "*.sql" | xargs grep -l "orders_mart"

8. Log Analysis Patterns

8.1 Structured Log Grep with jq

# Count errors by error type over time
cat application.log \
  | jq -r 'select(.level=="ERROR") | "\(.timestamp[:13]) \(.error_type // "unknown")"' \
  | sort | uniq -c | sort -rn

# Find first occurrence of a specific error
cat application.log \
  | jq 'select(.level=="ERROR" and (.message | contains("NullPointerException")))' \
  | jq -r '.timestamp' | sort | head -1

# Trace all log lines for a specific correlation ID
cat application.log \
  | jq -c 'select(.correlation_id == "abc-123-def")'

# Extract error messages and stack traces
cat application.log \
  | jq -r 'select(.level=="ERROR") | "\(.timestamp) | \(.message) | \(.exception // "")"'

8.2 Correlation ID Tracing Across Services

import json
from pathlib import Path
from collections import defaultdict

def trace_by_correlation_id(log_files: list[str], correlation_id: str) -> list[dict]:
    """Collect all log events for a given correlation_id across multiple log files."""
    events = []
    for log_file in log_files:
        for line in Path(log_file).read_text().splitlines():
            try:
                entry = json.loads(line)
                if entry.get("correlation_id") == correlation_id:
                    entry["_source_file"] = log_file
                    events.append(entry)
            except json.JSONDecodeError:
                continue
    return sorted(events, key=lambda e: e.get("timestamp", ""))


# Usage
events = trace_by_correlation_id(
    log_files=[
        "/var/log/airflow/orders_ingest.log",
        "/var/log/spark/orders_transform.log",
        "/var/log/dbt/orders_mart.log",
    ],
    correlation_id="run-2024-06-01-orders",
)
for ev in events:
    print(f"{ev['timestamp']} [{ev['level']}] {ev['_source_file']}: {ev['message']}")

8.3 Error Frequency Analysis — When Did It Start?

import json
from datetime import datetime
from collections import Counter
from pathlib import Path

def error_frequency_by_hour(log_file: str) -> None:
    """Count errors per hour to find when the issue started."""
    hourly_errors: Counter = Counter()
    hourly_total: Counter  = Counter()

    for line in Path(log_file).read_text().splitlines():
        try:
            entry = json.loads(line)
            ts = entry.get("timestamp", "")
            hour = ts[:13]  # "2024-06-01T14"
            hourly_total[hour] += 1
            if entry.get("level") in ("ERROR", "CRITICAL"):
                hourly_errors[hour] += 1
        except json.JSONDecodeError:
            continue

    print(f"{'Hour':<17} {'Total':>8} {'Errors':>8} {'Error%':>8}")
    print("-" * 45)
    for hour in sorted(hourly_total):
        total  = hourly_total[hour]
        errors = hourly_errors.get(hour, 0)
        pct    = 100.0 * errors / total if total else 0
        flag   = " <-- spike" if pct > 5 else ""
        print(f"{hour:<17} {total:>8} {errors:>8} {pct:>7.1f}%{flag}")

8.4 Airflow Log Aggregation with Pandas

import re
import pandas as pd
from pathlib import Path

LOG_PATTERN = re.compile(
    r"\[(?P<ts>[^\]]+)\] \{[^}]+\} (?P<logger>\S+) - (?P<level>\w+) - (?P<message>.+)"
)

def parse_airflow_logs(log_path: str) -> pd.DataFrame:
    records = []
    for line in Path(log_path).read_text().splitlines():
        m = LOG_PATTERN.match(line)
        if m:
            records.append(m.groupdict())
    df = pd.DataFrame(records)
    if df.empty:
        return df
    df["ts"] = pd.to_datetime(df["ts"], utc=True)
    return df

# Usage
df = parse_airflow_logs("/opt/airflow/logs/orders_etl/load_task/2024-06-01/1.log")

# Show only errors
print(df[df["level"].isin(["ERROR", "CRITICAL"])][["ts", "logger", "message"]])

# Find first error timestamp
first_error = df[df["level"] == "ERROR"]["ts"].min()
print(f"First error at: {first_error}")

9. RCA Document Template

Use this template immediately after resolving an incident. Fill it within 24 hours while memory is fresh.

# RCA: <Short Incident Title>

**Incident ID:** INC-YYYY-NNN
**Severity:**    P1 / P2 / P3
**Status:**      Resolved

---

## Incident Summary

| Field | Value |
|---|---|
| What broke | <one-line description of what failed> |
| Data affected | <tables, partitions, date ranges> |
| Users/systems impacted | <dashboards, reports, downstream pipelines> |
| Duration | <start datetime> → <end datetime> (<N hours>) |
| Detection method | Automated alert / user report / monitoring |

---

## Timeline (UTC)

| Time | Event |
|---|---|
| 2024-06-01 14:00 | Anomaly begins (root cause event) |
| 2024-06-01 15:42 | Alert fires / stakeholder reports issue |
| 2024-06-01 15:50 | On-call acknowledged, investigation started |
| 2024-06-01 16:15 | Root cause identified |
| 2024-06-01 16:45 | Fix deployed |
| 2024-06-01 17:00 | Data validated — incident resolved |
| 2024-06-01 17:30 | Stakeholders notified of resolution |

---

## Root Cause Statement

> <Single sentence: "X failed because Y, caused by Z.">

Example:
> The `orders_mart` model produced zero rows for 2024-06-01 because the incremental
> filter used `created_at` instead of `updated_at`, causing no new rows to be
> selected after a source migration changed the update timestamp field name.

---

## Contributing Factors

1. No data volume alert existed for `orders_mart` — zero rows went undetected for 90 minutes.
2. The source schema change was not communicated to the data engineering team.
3. The dbt model had no row-count test that would have failed the CI run.

---

## Evidence

- Airflow task logs: `gs://airflow-logs/orders_etl/orders_mart/2024-06-01/1.log`
- dbt run_results.json: `s3://artifacts/dbt/2024-06-01/run_results.json`
- Grafana dashboard showing row count drop: `https://grafana.internal/d/abc123`
- Source schema diff showing field rename: `https://github.com/org/source/pull/456`

---

## Fix Applied

```sql
-- Changed incremental filter from created_at to updated_at
-- File: models/marts/orders_mart.sql
{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

Full refresh run to repopulate the partition:

dbt run --select orders_mart --full-refresh --vars '{"start_date": "2024-06-01"}'

Prevention

---

## 10. Common Pitfalls in RCA

### Blaming the Person, Not the Process

An RCA that concludes "engineer X forgot to update the filter" is useless — it doesn't prevent recurrence. Correct framing: "the process had no automated check that would catch this class of error." Always ask: "What process, tool, or alert would have prevented a different engineer from making the same mistake?"

### Stopping at the First Cause

The first cause is almost always a symptom. Apply the **5-Whys** method:

1. Why did the dashboard show wrong revenue? → Because orders_mart had zero rows.

2. Why did orders_mart have zero rows? → Because the incremental filter matched nothing.

3. Why did the filter match nothing? → Because it used created_at > max(created_at in target), but the source only updates updated_at on modifications.

4. Why was the wrong column used? → Because the mod

…(truncated)