ETL Job Monitoring: Freshness, Row Counts, and Silent Failures

To monitor an ETL job, check what each run did to the data as well as whether it ran: rows read and loaded, rows rejected or duplicated, whether the target's totals match the source, and how old the newest row in the target is. Most ETL failures don't crash. A source that returns nothing, a load that stops halfway, and a retry that appends a second copy all exit 0. A monitor that expects every run catches runs that fail or never happen, numbers reported with each run catch the ones that succeed with bad data, and a separate freshness check catches a pipeline that runs on time on input that stopped changing.

The examples use PostgreSQL on both ends as a stand-in for an application database and a warehouse, Python 3.11+ with psycopg 3.1+, and dbt Core 1.12 for the transform step.

Failures That Exit 0#

• Zero rows. An API that answers 200 with an empty list, credentials for the wrong tenant, and a filter that stopped matching all load nothing, successfully. Cron job succeeded but did nothing covers where empty runs come from.
• A partial load. A loader that commits every 10,000 rows and dies at row 40,000 leaves a window that looks like a quiet hour, and a rerun that appends adds those 40,000 again.
• Duplicates from a retry. A load that timed out on the client after the server committed is retried and appends again. A primary key would refuse the copy, but Snowflake (standard tables), BigQuery (keys), and Amazon Redshift (constraints) don't enforce primary keys.
• Schema drift. A query that names a renamed column errors, which is the good case. Code that reads record.get("total_cents") from a JSON API loads NULLs after the rename, and a new field nobody maps is dropped. Google's SRE workbook describes that last case in a pipeline whose stages each report success: "Both jobs think they are correct, but the user doesn't see the data" (Data Processing Pipelines).
• Late-arriving rows. A row stamped 13:59 by a transaction that commits at 14:20 misses a 14:15 load of the 13:00 hour. Idempotent scheduled jobs explains why DEFAULT now() timestamps arrive late and covers upserts and watermarks.
• A stale source. A replica that stopped replicating, or a vendor file nobody replaced overnight, serves old data normally, and the job loads it again.
• The wrong window. Python's date.today() returns the local date (datetime), so a daily load at 00:30 UTC on a host set to New York computes "yesterday" as two days ago: every run succeeds, and the table is always a day behind. A naive datetime goes to PostgreSQL as a timestamp (psycopg), read in the session's time zone against timestamptz columns (date/time types), which shifts an hourly window by the gap between the two zones.

Measure Every Run#

Reconciliation compares source and target for the same window: row counts catch lost and doubled rows, and a control total such as the sum of an amount column also catches values truncated or mapped to the wrong column. When they disagree, per-hour counts on both sides show where.

A Load That Checks Itself Before It Commits#

This job reloads the last three closed UTC hours of orders into analytics.orders every hour, so rows that committed after the previous run read them still arrive. It streams the window with COPY (passing parameters to copy() needs psycopg 3.1, per its COPY docs), replaces the window instead of appending, reconciles it with the source inside the same transaction, and returns its numbers:

# load_orders.py (Python 3.11+, psycopg 3.1+): reload the last three closed UTC hours of
# orders into the warehouse, reconcile them with the source, and report the numbers.
import datetime as dt
import os
import time

import psycopg

from checkin import checkin

KEY = "orders-hourly-load"
LOOKBACK = dt.timedelta(hours=3)
COLUMNS = ["id", "customer_id", "total_cents", "status", "created_at"]
CONNECT = {"autocommit": True, "connect_timeout": 10, "options": "-c statement_timeout=15min"}


class LoadCheckFailed(Exception):
    """The load raised no error, but its data doesn't add up."""


def load_window(src, wh):
    end = dt.datetime.now(dt.UTC).replace(minute=0, second=0, microsecond=0)
    window = (end - LOOKBACK, end)  # aware datetimes, so no session time zone applies
    with src.transaction(), wh.transaction():
        # One snapshot for the counts and the copy, so both describe the same rows.
        src.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
        found = [c.name for c in src.execute("SELECT * FROM orders LIMIT 0").description]
        if found != COLUMNS:
            raise LoadCheckFailed(f"orders columns changed: {found}")
        src_rows, src_cents = src.execute(
            "SELECT count(*), coalesce(sum(total_cents) FILTER (WHERE customer_id IS NOT NULL"
            " AND total_cents >= 0), 0) FROM orders WHERE created_at >= %s AND created_at < %s",
            window,
        ).fetchone()
        newest = src.execute("SELECT max(created_at) FROM orders").fetchone()[0]
        if newest is None:
            raise LoadCheckFailed("orders is empty")

        wh.execute(
            "CREATE TEMP TABLE stage_orders (id bigint, customer_id bigint, total_cents bigint,"
            " status text, created_at timestamptz) ON COMMIT DROP"
        )
        copy_out = "COPY (SELECT * FROM orders WHERE created_at >= %s AND created_at < %s) TO STDOUT"
        with (
            src.cursor().copy(copy_out, window) as rows_out,
            wh.cursor().copy("COPY stage_orders FROM STDIN") as rows_in,
        ):
            for block in rows_out:
                rows_in.write(block)
        staged = wh.execute("SELECT count(*) FROM stage_orders").fetchone()[0]

        # Replace the window, so a rerun or retry can't append a second copy.
        wh.execute("DELETE FROM analytics.orders WHERE created_at >= %s AND created_at < %s", window)
        loaded = wh.execute(
            "INSERT INTO analytics.orders (id, customer_id, total_cents, status, created_at)"
            " SELECT id, customer_id, total_cents, status, created_at FROM stage_orders"
            " WHERE customer_id IS NOT NULL AND total_cents >= 0"
        ).rowcount
        tgt_rows, tgt_ids, tgt_cents = wh.execute(
            "SELECT count(*), count(DISTINCT id), coalesce(sum(total_cents), 0)"
            " FROM analytics.orders WHERE created_at >= %s AND created_at < %s",
            window,
        ).fetchone()

        # Raising here rolls back the delete and the insert; the previous load stays.
        if staged != src_rows or tgt_cents != src_cents:
            raise LoadCheckFailed(
                f"source {src_rows} rows, {src_cents} cents; staged {staged} rows;"
                f" target {tgt_rows} rows, {tgt_cents} cents"
            )
    return {
        "rows_extracted": staged,
        "rows_loaded": loaded,
        "rejected_rows": staged - loaded,
        "duplicate_keys": tgt_rows - tgt_ids,
        "source_lag_minutes": round((dt.datetime.now(dt.UTC) - newest).total_seconds() / 60, 1),
    }


if __name__ == "__main__":
    run_id = f"{KEY}-{int(time.time())}-{os.getpid()}"
    checkin(KEY, "start", run_id, schedule="15 * * * *", gracePeriod=600)
    try:
        with (
            psycopg.connect(os.environ["SOURCE_DSN"], **CONNECT) as src,
            psycopg.connect(os.environ["WAREHOUSE_DSN"], **CONNECT) as wh,
        ):
            outcomes = load_window(src, wh)
    except Exception as exc:
        # Database errors can quote row values, so send only their type.
        message = str(exc) if isinstance(exc, LoadCheckFailed) else type(exc).__name__
        checkin(KEY, "fail", run_id, message=message[:1000])
        raise
    checkin(KEY, "complete", run_id, body={"outcomes": outcomes})

• One snapshot. Under Read Committed, "two successive SELECT commands can see different data"; under Repeatable Read, every statement sees the first one's snapshot (transaction isolation), so an order committed mid-run can't make the counts and the copy disagree. The level must be set before the first query (SET TRANSACTION), hence autocommit: each transaction() block is then its own BEGIN and COMMIT, or rollback on an exception (transactions).
• Drift stops the run. SELECT * ... LIMIT 0 reads the column list, not rows. Any rename, removal, or addition fails the run, so a new field gets a decision instead of being dropped.
• Overwrite, don't append. The DELETE and INSERT commit together, so readers see the old window until the new one replaces it. On a warehouse, a MERGE on the business key or a partition overwrite does the same.
• Rejects are counted. Rows with no customer or a negative total stay out and show up as rejected_rows; the control total compares valid rows only.
• Nothing waits forever. statement_timeout aborts any statement past 15 minutes (client settings), turning a lock wait into a reported failure.

Here the source's primary key keeps duplicate_keys at 0; with files, an API, or change events delivered at least once, that count is what notices.

The job reports through this helper:

# checkin.py: CronRadar HTTP check-ins with a 5-second timeout. Errors are printed and
# ignored, so monitoring never changes the job's result.
import base64
import json
import os
import sys
import urllib.parse
import urllib.request


def checkin(key, event, run_id, body=None, **params):
    try:
        query = urllib.parse.urlencode({"run_id": run_id, **params}, quote_via=urllib.parse.quote)
        request = urllib.request.Request(
            f"https://cron.life/api/v1/ping/{urllib.parse.quote(key, safe='')}/{event}?{query}",
            # allow_nan=False refuses NaN and Infinity, which aren't valid JSON.
            data=json.dumps(body, allow_nan=False).encode() if body is not None else None,
            method="POST",
        )
        token = base64.b64encode(f"{os.environ['CRONRADAR_API_KEY']}:".encode()).decode()
        request.add_header("Authorization", f"Basic {token}")
        if body is not None:
            request.add_header("Content-Type", "application/json")
        urllib.request.urlopen(request, timeout=5).close()
    except Exception as exc:
        print(f"CronRadar {event} check-in failed: {exc}", file=sys.stderr)

urlopen raises on a 4xx or 5xx, and timeout=5 bounds the connection attempt and each blocking read (urllib.request); the except absorbs both, while the job still re-raises its own exception. Schedule it under a lock, with CRONRADAR_API_KEY, SOURCE_DSN, and WAREHOUSE_DSN in its environment:

15 * * * * flock --nonblock /var/lib/etl/orders-load.lock python3 /opt/etl/load_orders.py

flock --nonblock exits instead of waiting when the previous run still holds the lock (flock(1)), so two runs never load the same window at once.

Report the Numbers and Set the Expectations#

There is no installable CronRadar SDK for Python today and no auto-discovery for cron, so each job reports with its own HTTP check-ins, as above. The start carries the schedule (read in UTC), so the first run creates the monitor and a later run recreates a deleted one; schedule and gracePeriod apply only when a check-in creates the monitor, so change them later in its settings. A fail alerts immediately. A hung run is caught as stuck once the next expected run plus the grace period passes with a start as the last event; flock keeps the next run from sending a start that would reset that window, and statement_timeout usually turns the hang into a fail first.

The complete is a POST with a JSON body under outcomes: at most five values, keys that start with a lowercase letter and use only lowercase letters, digits, and underscores (64 characters at most), each a finite number or boolean. Configure expectations first; keys without one are accepted but not stored. Once one exists, a run that omits its key fails the check, as does a bare heartbeat. A missing or violated value marks the run outcome-failed and opens an incident showing observed and expected values. Never send rows, logs, secrets, or personal data; the API reference has the full rules.

After the first run, add these in the monitor's settings under Outcome checks:

Set floors at about half the quietest three hours in your history; these assume at least 2,000 orders in any three-hour window. Use a source lag limit only for a source that writes around the clock.

Test the body once by hand. CronRadar rejects an outcome body it can't accept, such as a sixth value, an uppercase key, or a string, with a 400 and doesn't record the completion. Since the next run's start still arrives on schedule, that can go on without an alert.

The start and complete pair also records each run's duration. There is no slow-run setting; to alert on a slow load, report duration_seconds against an expectation such as duration_seconds <= 1800, at the cost of one of the five values. Tracking cron job duration covers choosing the limit.

Check Freshness in Its Own Job#

Google's SRE workbook lists three formats for pipeline freshness SLOs: "X% of data processed in Y [seconds, days, minutes]", "The oldest data is no older than Y [seconds, days, minutes]", and "The pipeline job has completed successfully within Y [seconds, days, minutes]" (Data Processing Pipelines). A schedule monitor on the load answers the third. The first two are about the data, and only a check that reads the data answers them.

Run that check as a separate small job with its own monitor. It measures the table people read, whichever job writes it, and keeps running when the load is broken or replaced. It also catches the wrong-window bug: a load reloading hours from four hours ago reports normal counts and a fresh source, and only the target's newest row shows the problem.

-- /opt/etl/orders-freshness.sql: minutes since the newest order in analytics.orders,
-- and how many closed hours from 24 to 2 hours ago have no orders at all.
SELECT
  floor(extract(epoch FROM now() - max(created_at)) / 60)::bigint AS freshness_minutes,
  (SELECT count(*)
     FROM generate_series(date_trunc('hour', now(), 'UTC') - interval '24 hours',
                          date_trunc('hour', now(), 'UTC') - interval '2 hours',
                          interval '1 hour') AS h
    WHERE NOT EXISTS (SELECT 1 FROM analytics.orders
                       WHERE created_at >= h AND created_at < h + interval '1 hour')
  ) AS empty_hours
FROM analytics.orders;

empty_hours covers what freshness can't. If the host misses five hourly runs, the first run afterwards reloads only the last three hours: the newest row is fresh, and three earlier hours were never loaded. Missed runs after downtime covers what each scheduler does about skipped runs; a watermark of the last loaded hour lets the load catch up by itself. The three-argument date_trunc truncates in the named zone, not the session's (date/time functions). The check assumes every hour has orders; for a source with quiet hours, count longer buckets.

#!/bin/sh
# /opt/etl/check-orders-freshness.sh: report how far behind analytics.orders is.
key=orders-freshness
url="https://cron.life/api/v1/ping/$key"
run_id="$key-$(date +%s)-$$"

report() {
  event=$1
  shift
  curl --fail --silent --show-error --max-time 5 --output /dev/null \
    --request POST --get --user "$CRONRADAR_API_KEY:" \
    --data-urlencode "schedule=*/15 * * * *" \
    --data-urlencode "gracePeriod=300" \
    --data-urlencode "run_id=$run_id" \
    "$@" "$url/$event" || true
}

report start
# Bound the connection and the query, so a hung database becomes a failed check.
result=$(PGCONNECT_TIMEOUT=10 PGOPTIONS="-c statement_timeout=60s" \
  psql --no-psqlrc --tuples-only --no-align --set ON_ERROR_STOP=1 \
  --dbname="$WAREHOUSE_DSN" --file=/opt/etl/orders-freshness.sql)
rc=$?
if [ "$rc" -ne 0 ]; then
  report fail --data-urlencode "message=freshness query: psql exited with code $rc"
  exit "$rc"
fi
freshness=${result%%|*}
empty_hours=${result#*|}
if [ -z "$freshness" ]; then
  report fail --data-urlencode "message=analytics.orders has no rows"
  exit 1
fi

body=$(printf '{"outcomes":{"freshness_minutes":%d,"empty_hours":%d}}' "$freshness" "$empty_hours")
curl --fail --silent --show-error --max-time 5 --output /dev/null \
  --request POST --user "$CRONRADAR_API_KEY:" \
  --header "Content-Type: application/json" \
  --data "$body" "$url/complete?run_id=$run_id" || true

PGCONNECT_TIMEOUT and PGOPTIONS act as the connect_timeout and options connection parameters (libpq). psql exits 2 when the connection fails and 3 when a statement errors with ON_ERROR_STOP set, and prints NULL as nothing (psql), so an empty table leaves freshness blank. The JSON complete leaves out --get, which would move the body into the query string (curl).

Schedule it with */15 * * * * /opt/etl/check-orders-freshness.sh, then add freshness_minutes <= 90 and empty_hours = 0 under Outcome checks. A load at :15 that reloads up to the top of the hour keeps the newest row under about 80 minutes old, so a missed or misdirected load fails the check within about 30 minutes of when it should have finished.

Order Dependent Steps by Data, Not by the Clock#

A transform scheduled for :40 because the load usually finishes by :25 builds on old data, and succeeds, the day the load runs long. Either run the steps in one script that stops at the first failure, with one monitor, or make each downstream step check its input's freshness first and fail when it's stale. An orchestrator such as Apache Airflow runs dependent steps as tasks in one DAG; Airflow DAG schedule monitoring covers why a DAG run starts late or never starts.

dbt does the second with source freshness. From dbt 1.10, both freshness and loaded_at_field go under config (freshness):

# models/sources.yml
sources:
  - name: warehouse
    schema: analytics
    config:
      loaded_at_field: created_at
      freshness:
        warn_after: { count: 75, period: minute }
        error_after: { count: 90, period: minute }
    tables:
      - name: orders

In the dbt job's wrapper, between its start and its complete or fail:

dbt source freshness --select "source:warehouse.orders" && dbt build

dbt compares max(loaded_at_field) with the current time. In dbt Core 1.12, a source past error_after makes the command exit 1, while one past only warn_after logs a warning and exits 0 (runnable.py), so && skips the build only when the input is too old. dbt build also runs tests: at the default severity: error a failing test fails the invocation, while severity: warn only warns unless you pass --warn-error (severity). Either failure exits non-zero (exit codes), and the wrapper reports fail. dbt v2 adds dbt freshness, which also covers models, and keeps dbt source freshness working (docs, checked October 25, 2026). If pg_cron runs the transform inside the database, monitoring pg_cron jobs shows how to report its runs.

Frequently Asked Questions#

How do I monitor an ETL job?#

Report each run's start and its finish or failure to a monitor that expects the job on its schedule, so a missed, failed, or hung run alerts. With the finish, send counts such as rows extracted, loaded, and rejected, and alert when one is missing or out of range. Add a separate freshness check on the target table.

How do I detect an ETL job that loaded zero rows?#

Count the rows the load inserted, then fail the job below a floor or report the count with the run's completion and alert outside the job, as in rows_loaded >= 1000. Count rows extracted too, to tell an empty source from a load that dropped what it read.

What is data freshness monitoring?#

It measures how old the newest data in a table is, such as minutes since the newest order, and alerts past a limit. It catches a pipeline that succeeds on stale input or loads the wrong window, which job monitoring can't see, so run it as its own scheduled check.

How do I alert when a data pipeline is late?#

Give each scheduled job a monitor that knows its schedule and grace period, so a run that hasn't reported by then alerts. For late data, check the target's newest row on its own schedule and alert past a limit such as 90 minutes.

Should I use dbt source freshness or a heartbeat monitor?#

Both, because they answer different questions. dbt source freshness tells you whether a source is too old, but only when something runs dbt; if that job stops, nothing reports. A heartbeat or lifecycle monitor on the job tells you whether dbt ran on schedule and exited 0.