# ETL Job Monitoring: Freshness, Row Counts, and Silent Failures By Dan Ochoa · Published 2026-10-25 · Updated 2026-10-25 Canonical: https://cronradar.com/blog/etl-job-monitoring > Monitor ETL jobs past the exit code: row counts, rejects, duplicates, reconciliation, and a separate freshness check, with Python, SQL, and dbt examples. **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](https://cronradar.com/blog/scheduled-job-silent-success) 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](https://docs.snowflake.com/en/sql-reference/constraints-overview)), BigQuery ([keys](https://docs.cloud.google.com/bigquery/docs/primary-foreign-keys)), and Amazon Redshift ([constraints](https://docs.aws.amazon.com/redshift/latest/dg/t_Defining_constraints.html)) 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](https://sre.google/workbook/data-processing/)). - **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](https://cronradar.com/blog/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](https://docs.python.org/3/library/datetime.html#datetime.date.today)), 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](https://www.psycopg.org/psycopg3/docs/basic/adapt.html)), read in the session's time zone against `timestamptz` columns ([date/time types](https://www.postgresql.org/docs/current/datatype-datetime.html)), which shifts an hourly window by the gap between the two zones. ## Measure Every Run | Measure | How | Catches | | -------------- | ----------------------------------------------------- | ------------------------------------------ | | Rows extracted | Rows read from the source window | An empty or cut-off source | | Rows loaded | Row count of the insert into the target | A transform or filter that drops rows | | Rejected rows | Rows that failed validation | Drift that nulls a column, an upstream bug | | Duplicate keys | `count(*) - count(DISTINCT key)` in the target window | A retried or overlapping append | | Source lag | Age of the newest row in the source | A source that stopped changing | | Control total | Sum of a numeric column, source against target | Lost, doubled, or mangled values | | Duration | Start and finish times | A load growing toward its next run | | Bytes | Size of each file or extract | A truncated file | 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](https://www.psycopg.org/psycopg3/docs/basic/copy.html)), replaces the window instead of appending, reconciles it with the source inside the same transaction, and returns its numbers: ```python # 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](https://www.postgresql.org/docs/current/transaction-iso.html)), 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](https://www.postgresql.org/docs/current/sql-set-transaction.html)), hence autocommit: each `transaction()` block is then its own `BEGIN` and `COMMIT`, or rollback on an exception ([transactions](https://www.psycopg.org/psycopg3/docs/basic/transactions.html)). - **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](https://www.postgresql.org/docs/current/runtime-config-client.html)), 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: ```python # 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](https://docs.python.org/3/library/urllib.request.html#urllib.request.urlopen)); 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: ```text 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)](https://man7.org/linux/man-pages/man1/flock.1.html)), 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](https://docs.cronradar.com/api-reference) has the full rules. After the first run, add these in the monitor's settings under Outcome checks: | Outcome check | Catches | | -------------------------- | -------------------------------------------- | | `rows_extracted >= 1000` | An empty or cut-off source window | | `rows_loaded >= 1000` | Rows read but not loaded | | `rejected_rows <= 10` | A rename or upstream bug that nulls a column | | `duplicate_keys = 0` | A second copy of any order in the window | | `source_lag_minutes <= 30` | A source that stopped taking writes | 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](https://cronradar.com/blog/cron-job-duration-tracking) 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](https://sre.google/workbook/data-processing/)). 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. ```sql -- /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](https://cronradar.com/blog/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](https://www.postgresql.org/docs/current/functions-datetime.html)). The check assumes every hour has orders; for a source with quiet hours, count longer buckets. ```bash #!/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](https://www.postgresql.org/docs/current/libpq-envars.html)). psql exits 2 when the connection fails and 3 when a statement errors with `ON_ERROR_STOP` set, and prints NULL as nothing ([psql](https://www.postgresql.org/docs/current/app-psql.html)), so an empty table leaves `freshness` blank. The JSON `complete` leaves out `--get`, which would move the body into the query string ([curl](https://curl.se/docs/manpage.html)). 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](https://cronradar.com/blog/airflow-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](https://docs.getdbt.com/reference/resource-properties/freshness)): ```yaml # 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`: ```bash 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](https://github.com/dbt-labs/dbt-core/blob/v1.12.6/core/dbt/task/runnable.py#L803-L820)), 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](https://docs.getdbt.com/reference/resource-configs/severity)). Either failure exits non-zero ([exit codes](https://docs.getdbt.com/reference/exit-codes)), and the wrapper reports `fail`. dbt v2 adds `dbt freshness`, which also covers models, and keeps `dbt source freshness` working ([docs](https://docs.getdbt.com/reference/commands/source), checked October 25, 2026). If pg_cron runs the transform inside the database, [monitoring pg_cron jobs](https://cronradar.com/blog/pg-cron-monitoring) 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.