If your pipeline breaks at 3 a.m. and the only signal you have is "job failed," you don't have monitoring - you have vibes. Monitoring is how you move from "something is wrong" to "I know exactly where and why it broke" before stakeholders start pinging you about missing dashboards. This is a practical guide to monitoring as a data engineer: metrics, logs, and traces - what they are, how they fit together, and how to use them without drowning in noise.

1. Why monitoring matters for data engineers

Modern data stacks are sprawling:

• ELT/ETL pipelines across Airflow / Dagster / Prefect
• Warehouses (Snowflake, BigQuery, Redshift)
• Lakes (S3, ADLS, GCS)
• Streaming (Kafka, Kinesis)
• Microservices and APIs feeding you data

When something breaks, the business impact is immediate: a wrong finance report drives bad leadership decisions, a model trained on stale data degrades recommendations, and a missed partner SLA means penalties or lost trust.

Monitoring for data engineers boils down to answering three questions quickly:

1. Is the system healthy? (metrics)
2. What exactly is happening? (logs)
3. How did this request/job flow through the system? (traces)

Wire these three right and you move from "panic and grep" to "observe, pinpoint, fix."

2. Metrics, logs, traces - the core concepts

Treat observability as three complementary camera angles on the same system.

2.1 Metrics - "How healthy is it?"

Metrics are numeric time-series - values sampled over time. For data platforms, typical metrics:

• Pipeline level: pipeline_run_count, pipeline_failure_rate, pipeline_latency_seconds (end-to-end), sla_breaches_total
• Data quality: row_count, null_ratio_per_column, duplicate_keys_count
• Infra: CPU / memory / disk / I/O, connection pool usage, query latency

Metrics are cheap to store (aggregated, pre-computed) and great for dashboards and alerts, but poor at explaining why something broke - they show symptoms, not causes.

2.2 Logs - "What exactly happened?"

Logs are events over time, ideally structured. For data engineers, logs should tell you which job/task ran, which dataset/table was touched, which upstream dependency or external system failed, and which error or exception occurred.

Bad logs say "Error occurred." Good logs are structured JSON:

{
  "timestamp": "2025-11-26T02:14:03Z",
  "pipeline": "daily_orders_etl",
  "task": "load_to_warehouse",
  "run_id": "2025-11-26T02:00:00Z",
  "status": "FAILED",
  "error_type": "SnowflakeQueryError",
  "error_message": "Table ORDERS_RAW does not exist",
  "rows_processed": 0
}

Logs are detailed enough to explain why, but expensive if unstructured and noisy, and hard to query if you don't enforce consistency.

2.3 Traces - "How did this request/job flow?"

Traces show the path of a single thing (request, job, event) through multiple systems. For data engineers, that means an API call → writes to Kafka → consumed by a Spark job → written to Snowflake → triggers a dbt model, or a scheduled DAG run with each task fanning out to downstream services and queries.

A high-level trace:

• trace_id: abcd-1234
  • span: api_ingest_order (100 ms)
  • span: kafka_publish (50 ms)
  • span: spark_transform_orders (120s)
  • span: snowflake_load_orders_curated (30s)

Traces are gold for latency debugging and dependency analysis, underused in data teams compared to metrics and logs, and best implemented with a standard like OpenTelemetry.

3. A simple monitoring architecture for a data platform

Take a basic stack: an orchestrator (Airflow / Dagster / Prefect), a warehouse (Snowflake / BigQuery / Redshift), S3 for storage, and Spark / dbt for transforms. A pragmatic monitoring architecture:

1. Emit metrics - from the orchestrator (task success/failure, duration, backlog), from jobs (row counts, null ratios, validation failures), and from the warehouse (query latency, queueing, credits usage where available).
2. Collect logs - standardize on JSON, centralize in a log platform (Elasticsearch/OpenSearch, Loki, or cloud-native logging), and use consistent fields (pipeline, task, run_id, dataset, status).
3. Add traces where feasible - label everything with trace_id / run_id / correlation_id, use OpenTelemetry to propagate IDs, and trace a DAG run or streaming consumer end-to-end.
4. Visualize - metrics to Grafana or a BI dashboard, logs to Kibana/Grafana/cloud console, traces to Jaeger/Tempo/Zipkin or a vendor APM.

4. Concrete example: instrumenting a batch pipeline

Say you have a daily pipeline: extract from API → S3, load S3 → Snowflake stage table, transform stage → curated table.

4.1 Metrics to track

Pipeline-level:

• pipeline_run_duration_seconds{pipeline="daily_orders"}
• pipeline_run_status{pipeline="daily_orders", status="success|fail"}
• rows_loaded{table="orders_stage"}
• rows_loaded{table="orders_curated"}

Data quality:

• null_ratio{table="orders_curated", column="customer_id"}
• duplicate_key_count{table="orders_curated", column="order_id"}

Warehouse (if available): warehouse_query_duration_seconds, warehouse_queue_time_seconds, credits_used per job.

Pseudo-Python (Prometheus style):

from prometheus_client import Counter, Histogram, Gauge
import time

pipeline_duration = Histogram(
    "pipeline_run_duration_seconds",
    "Duration of pipeline run",
    ["pipeline"]
)

pipeline_status = Counter(
    "pipeline_run_status_total",
    "Pipeline run status",
    ["pipeline", "status"]
)

row_count_gauge = Gauge(
    "table_row_count",
    "Row count per table",
    ["pipeline", "table"]
)

def run_daily_orders():
    start = time.time()
    pipeline = "daily_orders"

    try:
        rows_stage = load_orders_stage()
        row_count_gauge.labels(pipeline=pipeline, table="orders_stage").set(rows_stage)

        rows_curated = transform_orders_curated()
        row_count_gauge.labels(pipeline=pipeline, table="orders_curated").set(rows_curated)

        pipeline_status.labels(pipeline=pipeline, status="success").inc()
    except Exception:
        pipeline_status.labels(pipeline=pipeline, status="fail").inc()
        raise
    finally:
        duration = time.time() - start
        pipeline_duration.labels(pipeline=pipeline).observe(duration)

Bare-bones, but already enough to see failures per day, watch duration creep over time, and compare row counts between steps.

5. Logging for data pipelines without going insane

5.1 Logging principles

1. Make logs structured - JSON with consistent keys: pipeline, task, dataset, run_id, trace_id, status, error_type.
2. Log lifecycle stages - STARTED, IN_PROGRESS, COMPLETED, FAILED, per task and per dataset touch.
3. Don't log raw sensitive data - no full records with PII, no access tokens, DB passwords, or secrets.
4. Log context, not noise - key IDs, counts, boundaries, error summaries; avoid spamming debug logs in tight loops.

5.2 Example: good logging for a load task

import logging
import json
from datetime import datetime

logger = logging.getLogger("daily_orders")

def log_event(**kwargs):
    # simple helper to enforce structured logs
    base = {
        "timestamp": datetime.utcnow().isoformat(),
        "service": "data-pipelines",
    }
    base.update(kwargs)
    logger.info(json.dumps(base))

def load_to_snowflake(run_id: str, source_path: str, table: str):
    log_event(
        level="INFO",
        pipeline="daily_orders",
        task="load_to_snowflake",
        run_id=run_id,
        event="STARTED",
        source_path=source_path,
        target_table=table,
    )

    try:
        rows = do_copy_into_snowflake(source_path, table)

        log_event(
            level="INFO",
            pipeline="daily_orders",
            task="load_to_snowflake",
            run_id=run_id,
            event="COMPLETED",
            rows_loaded=rows,
        )
    except Exception as e:
        log_event(
            level="ERROR",
            pipeline="daily_orders",
            task="load_to_snowflake",
            run_id=run_id,
            event="FAILED",
            error_type=type(e).__name__,
            error_message=str(e)[:500],
        )
        raise

Now your log search can query pipeline:daily_orders AND task:load_to_snowflake AND event:FAILED instead of playing regex roulette on random text.

6. Traces for data workflows: when and how

Traces are usually introduced by platform/DevOps teams, but data engineers benefit massively from them.

6.1 Where traces shine

• Complex DAGs with many dynamic tasks
• Systems mixing streaming consumption, microservices, and batch pipelines
• SLA-critical flows where you care where time is spent

Example: an order placed in the app should be visible in analytics within 2 minutes. Trace the flow:

1. API create_order → emits an event with a trace_id
2. Kafka produces the orders topic carrying the same trace_id
3. Spark streaming job consumes; spans wrap each micro-batch
4. Warehouse load: a final span logging the write to the analytics table

When the SLA breaks, you can answer: is it stuck at the API, the topic, the consumer, or the warehouse?

6.2 A minimal tracing mindset

Even without Jaeger or Tempo, you can generate a trace_id / run_id at the start of a pipeline, then pass it through logs, metrics labels, and warehouse audit columns (trace_id, run_id) to correlate events manually:

import uuid

def start_pipeline():
    run_id = uuid.uuid4().hex
    # use run_id in metrics labels, logs, and db audit columns
    return run_id

7. Best practices and common pitfalls

7.1 Best practices

1. Start with SLOs, not tools. Decide what you care about - "95% of daily pipelines finish before 7:00 AM," "streaming lag under 60 seconds for 99% of events," "no more than 1% of records fail validation" - then define metrics and alerts to match.

2. Standardize observability fields. Use the same naming everywhere - pipeline, task, job, dataset, run_id, trace_id, status, env - and agree on them across teams.

3. Instrument only what you'll use. Start with pipeline duration, success/failure, and row counts per key table; add more after you've proven you actually look at the dashboards.

4. Use sampling. Log full details on failures but sample successful runs; sample a small percentage of traces for high-volume paths.

5. Make alerts actionable. "High CPU" and "pipeline failed" with no context are useless. Prefer "daily_orders pipeline failed 3 times in the last hour in prod" and "row count for orders_curated dropped >20% vs the 7-day average."

7.2 Common pitfalls

1. Only monitoring infra, not data. CPU can be fine while a table has 0 rows. Always include data-level metrics: row counts, quality checks.
2. No correlation IDs. If logs, metrics, and tables don't share a run_id, connecting the dots is impossible.
3. Log noise. Debug logs flooding every second make the platform unusable - fix with log levels, structured logs, and aggregation.
4. "Set it and forget it" dashboards. Treat dashboards like code: review and refactor as systems evolve.
5. No drill-down path. Design the flow from metric → relevant logs → related traces → warehouse records intentionally.

8. A practical monitoring checklist

For each critical pipeline, aim for this minimal baseline:

• Metric: run duration
• Metric: success/failure (labels pipeline, env)
• Metric: key row counts (input and output tables)
• Log: structured logs with pipeline, task, run_id, status
• Log: errors with error_type and truncated error_message
• Optional: trace/run ID propagated into tables as audit columns

Platform-level:

• Dashboard: pipeline success rate over time
• Dashboard: pipeline duration trends
• Dashboard: data volume trends (row counts) for key tables
• Alerts on: repeated failures for the same pipeline, large deviations in row counts or data quality, and SLA breaches on critical jobs

Metrics give you the health overview, logs give you the story and root cause, and traces give you the end-to-end journey. Start painfully simple - instrument your most critical pipelines with run duration, success/failure, and row counts, add structured logs with a run_id, and bring in tracing once streaming and microservices make the flow complex. You can always add more; what kills teams is never starting, or building a noisy, unusable mess.