7 posts tagged with "Airflow"

The alert email: production server root partition at 81% (30G/40G, past the 80% red line). First instinct: check docker system df — and there it is, images at RECLAIMABLE 100% (7), apparently a quick win. Except all 7 images are running; following the hint with docker image prune -a would be a production incident.

Encountered this while building AI Analytics — an LLM-powered analytics platform that surfaces market trends, user behavior, and sales data; the disk in question belongs to the Docker server running its Airflow data pipeline.

TL;DR​

Past a disk red line, don't trust docker system df RECLAIMABLE — it's an estimate of "space with no container reference", and active images get flagged 100% anyway. The right move: sudo du -xh -d1 / layer by layer. This incident's three invisible hogs were none of them business data: task logs accumulating in container writable layers (no log volume in the Dockerfile), package-manager caches (npm + pnpm, ~4G combined), and a backup script that never pruned (full clones every run). Fixes: delete caches, reclaim writable layers with force-recreate, add TTL pruning to the backup script — 81% back down to 64%.

Symptoms​

$ df -h /
Filesystem      Size  Used Use% Mounted on
/dev/vda1        40G   30G  81% /

$ docker system df
TYPE        TOTAL   ACTIVE   SIZE     RECLAIMABLE
Images      7       7        4.2GB    100% (7)    ← all running
Containers  5       5        810MB    0%

Acting on docker system df (clean images) is a dead end — RECLAIMABLE says 100% but ACTIVE says 7/7. Where the space actually went, docker system df never shows:

sudo du -xh -d1 / | sort -rh | head
# selected output:
# 2.7G    /root/.npm              ← npm cache
# 1.1G    /root/.local/share/pnpm ← pnpm store
# 857M    /root/backups-git       ← backup clones, never pruned
# (hidden in overlay2: airflow scheduler writable layer 468M + dag-processor 257M)

Root Cause​

Three kinds of consumption, all invisible from the "business data" perspective.

Writable layers eating logs. Airflow task logs are written inside the containers with no external volume — scheduler writable layer 468M, dag-processor 257M, ~780M accumulated in two weeks, monotonically growing. Image layers never change; the writable layer does. docker system df buries it in the Containers SIZE total (810M), where it has no presence.

Package-manager caches only grow. On a server with frequent deploys/builds, ~/.npm (2.7G) and the pnpm store (1.1G) pile up indefinitely; nobody ever cleans them.

A backup script that never prunes. The backup script clones the repo worktree in full and pushes to GitHub every run — old clone directories are never removed, so 857M is mostly historical duplicates. Sneakier: orphan clones whose branch never pushed successfully can't just be deleted; verify first.

And docker system df RECLAIMABLE is a statistical estimate of "space not referenced by a running container" — active images can display 100% reclaimable (all 7 of ours did). It's a false-positive generator, not a cleanup guide.

Solution​

Step 1: Locate with du, layer by layer — look before deleting​

sudo du -xh -d1 / | sort -rh | head        # root partition, level by level
sudo du -xh -d1 /var/lib/docker | sort -rh | head   # drill into Docker's dir
docker ps -as --format "table {{.Names}}\t{{.Size}}"  # per-container writable layer

du -x stays on one filesystem, avoiding /proc, /sys noise and overlay confusion; docker ps -as SIZE exposes each container's writable layer — the key command for the "logs written inside the container" family.

Step 2: Delete caches outright​

npm cache clean --force        # or simply rm -rf ~/.npm/_cacache
pnpm store prune               # removes only unreferenced packages

~5.1G reclaimed, zero risk — caches re-download on demand.

Step 3: Reclaim writable layers with force-recreate​

docker compose up -d --force-recreate   # writable layer goes away with the old container

~780M reclaimed here. Two preconditions: pick a low-traffic window (services restart), and rescue anything valuable first — e.g. docker cp the task logs out, or they vanish with the container.

Step 4: TTL for backups, prevent recurrence​

# keep backup dirs for 7 days, prune older ones
find /root/backups-git -maxdepth 1 -type d -mtime +7 -exec rm -rf {} +

Add the pruning to the tail of the backup script (deployed in our github-backup-push.sh); before deleting, check for orphan clones whose branch never pushed (git ls-remote to compare) and confirm no unique commits — our 325M of orphans were removed only after verification.

After cleanup: 81% → 64%, with caches + pruning + writable layers together reclaiming ~6.7G.

FAQ​

docker system df shows 100% RECLAIMABLE — can I just delete?​

No. RECLAIMABLE estimates "space with no container reference", and running active images can still be flagged 100% reclaimable — all 7 of ours were. It answers "how much is theoretically unreferenced", not "what can be deleted". Measure first with du -xh -d1 and docker ps -as.

What's the difference between docker ps SIZE and docker system df SIZE?​

docker ps -as SIZE is the per-container writable layer; docker system df is the category-level aggregation over images/containers/volumes/cache. Logs accumulating inside a container only ever appear in the writable layer (visible via docker ps -as), never in any image size — which is exactly why "images look small but disk keeps growing".

How do I clean up a Docker server running out of disk?​

Three categories: package-manager caches deleted outright (npm cache clean --force, pnpm store prune), zero risk; container writable layers reclaimed via docker compose up -d --force-recreate (service restart; rescue logs first); backups, logs, and clone directories put on TTL pruning to prevent recurrence. Before any deletion, confirm the hogs with du -xh -d1 /.

Auditing a production task's decision snapshot, we found stage 5's feature data sitting in array slot 3 instead of the documented slot 4. Downstream consumers and the detail drawer read by "segment number minus one" — and were silently getting the previous stage's output.

Encountered this while building AI Analytics — an LLM-powered analytics platform that surfaces market trends, user behavior, and sales data; the snapshot is what the pipeline leaves behind for frontend rendering and post-hoc audit.

TL;DR​

The pipeline appends each stage's output into the snapshot array in execution order, and an if rows.empty: skip guard drops 0-row segments entirely, shifting every later segment forward. The static mapping "segment number − 1 = slot" breaks at will, and which segments are empty depends on runtime state — slots differ per run. Two fixes: resolve slots at read time by in-row keys (recommended), or keep placeholders for empty segments so slots stay constant.

Symptoms​

The assembly logic looks like this:

snapshot = {"features": [], "rule_output": []}
for seg in segments:                    # stages 1..5 run in order
    df = execute_sql(seg.sql)
    if not df.empty:                    # 0-row segments dropped here
        snapshot["features"].append(df.to_dict("records"))

The design assumption was "stage 5 → slot 4". Production snapshot audit found stage 5's features in slot 3:

all segments populated:   ①→0  ②→1  ③→2  ④→3  ⑤→4   ✓ matches assumption
stage 4 empty, dropped:   ①→0  ②→1  ③→2       ⑤→3   ✗ shifted
stages 3+4 empty:         ①→0  ②→1               ⑤→2   ✗ shifted again

The same code version produces completely different slot layouts depending on shop and permissions — an empty whitelist table drops stage 4; a disabled feature flag drops stage 3. Slots drift with runtime state.

Root Cause​

Positional addressing collided with sparse assembly. The snapshot array is a runtime product of concatenating "segments that produced output" — a sparsely-filled collection compressed. Consumers hard-coding "segment − 1" implicitly assume every segment always yields at least one row. That assumption shatters in three routine situations: empty whitelist tables, disabled feature flags, naturally empty business data. 0-row segments are the norm, not the exception.

Deeper down, the emptiness guard itself (if not df.empty) is not wrong — wrong is the contract's implicit premise. The design doc says "slot = segment − 1" but nobody declared it as an explicit contract. Every position-addressing consumer inherits an unacknowledged, unmaintained assumption.

Solution​

Option A (recommended): resolve segments by in-row keys at read time​

Let every row carry a segment identifier key; consumers resolve positions at read time with no static mapping:

def locate_segment(features: list, seg_key: str) -> dict:
    for row in features:
        if seg_key in row:              # row carries its own identity
            return row
    raise KeyError(f"segment '{seg_key}' missing in snapshot")

Slot drift becomes irrelevant — you look for "the segment whose keys look like this", not "element N". The one requirement: all consumers go through this single resolution entry point (write it into the processor docstring and the consumer contract: no hard-coded slots, ever).

Option B: keep placeholders for empty segments on the write side​

If downstream can't change yet, keep "slot = segment number" constant at assembly time:

snapshot["features"].append(
    df.to_dict("records") if not df.empty else {"__empty__": True}
)

The cost: placeholder objects appear in the snapshot and every consumer must handle them. Fine as a transition; converge on Option A long-term.

Step 3: contract tests over multiple runtime states​

Build snapshot fixtures for "all populated / one empty / several empty" and assert consumers parse all three identically. Testing only the all-full scenario tests nothing.

FAQ​

How do you handle schema drift in data engineering?​

Replace positional contracts with key-based ones: address snapshots, messages, and interfaces by field name or segment identifier, never array index. When upstream changes without notice (empty segments skipped, fields added or removed), key-addressed consumers at worst raise "not found" — they never silently read the wrong data.

What is the difference between schema drift and schema evolution?​

Evolution is explicitly managed versioning: add a field, cut a release, migrate consumers. Drift is passive: upstream changes and downstream quietly misaligns. The slot shifting in this post is textbook drift — nobody touched the contract; the data shape changed.

How do I detect this kind of slot misalignment in a pipeline?​

Two layers: contract tests covering multiple runtime states (all populated / one empty / several empty) asserting identical consumer parsing; and periodic production snapshot audits checking each slot's content against its declared segment. "Content and position disagree" is drift.

While triggering the same DAG repeatedly via the Airflow REST API for a staged rollout check, the request came back 4xx with no dag_run_id in the body — the DAG never actually ran — yet the script treated it as success.

Encountered this while building AI Analytics — LLM-powered analytics that surfaces market trends, user behavior, and sales data for precise operations strategy. The staged rollout of the ad-decision pipeline needed to trigger the same analysis repeatedly on Airflow for comparison, and some triggers were failing silently.

TL;DR​

Airflow enforces a unique constraint on each DAG's logical_date (dag_run_id must also be unique). POSTing /dags/{dag_id}/dagRuns with a logical_date that already exists gets rejected with a 4xx, and the response body contains no dag_run_id. If you only check the HTTP status code and don't inspect the returned dag_run_id, you'll mistake the rejection for success. Fix: use a distinct logical_date (and dag_run_id) on every trigger.

Symptoms​

To run a comparison test, the same DAG was triggered repeatedly with a fixed date 2026-01-01:

$ curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "dag_run_id": "manual-run-1",
      "logical_date": "2026-01-01T00:00:00Z"
    }'
# First time: returns a normal dag_run object with dag_run_id ✅

$ curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "dag_run_id": "manual-run-2",
      "logical_date": "2026-01-01T00:00:00Z"   # ⚠️ same logical_date
    }'
# Second time: returns an error object, no dag_run_id ❌
{
  "detail": "...",
  "status": 400,
  "title": "Bad Request",
  "type": "https://airflow.apache.org/docs/apache-airflow/2/stable-rest-api-ref.html#/default/Error"
}

If the caller only checks "is it 2xx" and stops there, or parses the JSON without verifying that dag_run_id exists, the second failure is silently swallowed — no error in the logs, no run in the Airflow UI.

Root Cause​

Airflow uses dag_run_id as the primary key for each run and maintains uniqueness on (dag_id, logical_date) in the metadata DB's dag_run table. logical_date is the "logical time" of a run — the scheduler uses it to decide whether a given schedule slot has already executed. Once a run with some logical_date exists for a DAG, triggering again with the same value is rejected to prevent duplicate execution.

The catch is that this failure is a 4xx with an error JSON, not a connection error or a 5xx. Many scripts only do a coarse response.status_code == 200 check, or grab the JSON and read fields without verifying dag_run_id is present — so "creation rejected" reads as "creation succeeded".

Solution​

Core idea: use a distinct logical_date (and dag_run_id) on every trigger. For replay / rollout-comparison scenarios, just append a counter to the date:

# Each iteration uses a different logical_date (2026-01-01 / 02 / 03 …)
for i in 1 2 3; do
  curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d "{
      \"dag_run_id\": \"manual-run-$i\",
      \"logical_date\": \"2026-01-0${i}T00:00:00Z\"
    }"
done

Even better, use an incrementing timestamp so logical_date and dag_run_id never collide. More importantly: always verify the dag_run_id field in the response — treat it as the only proof the trigger actually succeeded:

import requests

def trigger_dag(dag_id: str, logical_date: str, conf: dict | None = None) -> str:
    resp = requests.post(
        f"{AIRFLOW}/api/v2/dags/{dag_id}/dagRuns",
        headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
        json={"dag_run_id": f"manual-{logical_date}", "logical_date": logical_date, "conf": conf or {}},
    )
    # ❌ Not enough: status-only check lets 4xx slip through as success
    # resp.raise_for_status()
    data = resp.json()
    # ✅ Correct: only count it as created if dag_run_id is present
    if "dag_run_id" not in data:
        raise RuntimeError(f"Trigger failed: {resp.status_code} {data}")
    return data["dag_run_id"]

# A distinct logical_date each time makes repeated triggers safe
for i in range(1, 4):
    trigger_dag("my_dag", f"2026-01-0{i}T00:00:00Z")

Keep dag_run_id unique too — it's the primary key, and duplicates are rejected outright. A "prefix + logical_date" convention is common: unique, and easy to spot in the UI.

FAQ​

How do you trigger a DAG with the Airflow REST API?​

POST /api/v2/dags/{dag_id}/dagRuns with a body containing at least dag_run_id and logical_date (plus an optional conf for parameters). Both must be unique within the same DAG, or Airflow returns 4xx. In code, prefer the TriggerDagRunOperator, which also generates a unique run id internally.

Why does triggering the same DAG repeatedly fail in Airflow?​

Because Airflow maintains a unique constraint on (dag_id, logical_date) in the dag_run table, and dag_run_id itself is a primary key. Duplicate logical_date or dag_run_id values are rejected with 4xx. For replays or staged comparisons, give each trigger a fresh logical_date (or incrementing timestamp).

After deleting a DAG's .py file in Airflow to retire it, the dag_id still hangs around in the Web UI list and the database; even stranger — if you clear the metadata first and delete the file second, the just-cleared rows "come back to life."

Encountered this while building AI Ops — an LLM-powered analytics pipeline where retiring an old report DAG required cleaning its metadata too, otherwise the UI list and scheduled scans stayed polluted by residual rows.

TL;DR​

Airflow's dag-processor periodically scans the DAG folder and re-registers DAGs, and airflow dags reserialize doesn't purge "file-already-deleted" orphan rows — so just deleting the .py file won't make the dag_id vanish from the UI or DB. Conversely, clearing metadata before deleting the file lets the processor re-register the cleared rows on its next scan ("revival"). Correct order: ①delete the file first so the processor stops registering → ②SQL DELETE in foreign-key order → ③run airflow dags reserialize to verify.

Symptoms​

Retiring the shop_report_aggregation DAG — after deleting its .py file:

$ ls /opt/airflow/project/airflow_dags/shop_report_aggregation.py
ls: cannot access '.../shop_report_aggregation.py': No such file or directory

$ # but it's still in the database
$ docker exec cclhub-db psql -U airflow -d airflow -c \
    "SELECT dag_id, is_paused, is_active FROM dag WHERE dag_id='shop_report_aggregation';"
       dag_id            | is_paused | is_active
--------------------------+-----------+-----------
 shop_report_aggregation | f         | t          ← still there

It's not just the dag table — the matching rows in serialized_dag, dag_code, and dag_version are all still there, so the Web UI keeps showing this "deleted" DAG.

Worse is the reverse order — clear metadata first, delete file second:

T0  DELETE FROM dag WHERE dag_id='shop_report_aggregation';   ← cleared
T1  (.py file not deleted yet)
T2  dag-processor scan fires; file exists, dag table has no row → re-registers
T3  SELECT ... FROM dag WHERE dag_id='shop_report_aggregation';   ← it's back (revival)

Root Cause​

Two mechanisms stack up:

1. dag-processor scans and re-registers periodically. Airflow's dag-processor (part of the Scheduler) scans dags_folder on processor_poll_interval (default ~5 min), parses each .py file, and upserts into the metadata tables (dag, serialized_dag, dag_version). As long as the file exists, the next scan rewrites those rows. That's the direct source of "revival" — you clear the row, the file is still there, and the processor re-registers it as a new DAG.

2. reserialize ignores "file-gone" orphan rows. airflow dags reserialize re-serializes existing DAG files and refreshes serialized_dag; it does not delete orphan dag rows whose files have vanished. And airflow dags cleanup only purges expired dag_run history by default — it also leaves the dag / serialized_dag / dag_code / dag_version metadata tables alone. So after you delete the file, the metadata rows become orphans nobody cleans.

┌─ dag-processor ──────────────────────────────┐
│  scans dags_folder                            │
│  ├─ file present → upsert dag / serialized... │  ← source of revival
│  └─ file absent  → skip, no row deletion      │  ← orphan residue
└───────────────────────────────────────────────┘

Conclusion: to actually remove the metadata, you must give the processor no file to register (delete the file first), then manually clean the residual rows.

Solution​

Step 1: Delete the file first​

Make the .py file disappear from the DAG folder so dag-processor stops registering it.

# In production this is usually synced to the volume-mounted DAG folder
# /opt/airflow/project/airflow_dags/ via git pull
git pull   # removes shop_report_aggregation.py from the repo and the folder

# or delete directly (after confirming nothing depends on it)
rm /opt/airflow/project/airflow_dags/shop_report_aggregation.py

Step 2: Clean metadata in foreign-key order​

DELETE in foreign-key dependency order to avoid constraint violations. Deleting dag_run CASCADEs to task_instance:

BEGIN;

-- 1. run history (CASCADEs to task_instance)
DELETE FROM dag_run      WHERE dag_id = 'shop_report_aggregation';

-- 2. serialized DAG
DELETE FROM serialized_dag WHERE dag_id = 'shop_report_aggregation';

-- 3. version
DELETE FROM dag_version  WHERE dag_id = 'shop_report_aggregation';

-- 4. dag main table
DELETE FROM dag          WHERE dag_id = 'shop_report_aggregation';

-- 5. dag_code is keyed by source hash; multiple DAGs may share the same code;
--    only delete hashes no longer referenced by any serialized_dag
DELETE FROM dag_code
WHERE dag_hash NOT IN (SELECT dag_hash FROM serialized_dag);

COMMIT;

Step 3: Verify​

airflow dags reserialize

# confirm the dag row is not rebuilt
docker exec cclhub-db psql -U airflow -d airflow -c \
  "SELECT count(*) FROM dag WHERE dag_id='shop_report_aggregation';"
#  count
# -------
#      0   ✅

After reserialize, dag / serialized_dag / dag_code / dag_version are all 0 for that dag_id, and the next processor scan doesn't rebuild them — the cleanup is stable.

As a side note, on the same pipeline, pandas NaN crashing XCom serialization is another pitfall worth bookmarking.

Notes​

FAQ​

Why does a DAG still show in Airflow after deleting its .py file?​

Deleting the file doesn't clean the database. Rows in dag / serialized_dag / dag_code / dag_version still exist, and the Web UI reads those tables to render the list, so the deleted DAG keeps showing. Airflow has no built-in command to purge these orphan rows automatically; you must SQL DELETE them manually in foreign-key order.

How do I completely delete an Airflow DAG and all its metadata?​

Three steps: 1) delete the .py file so dag-processor stops registering it; 2) SQL DELETE in foreign-key order (dag_run → serialized_dag → dag_version → dag → orphan dag_code); 3) run airflow dags reserialize, then query the dag table to confirm the dag_id row count stays at 0 and isn't rebuilt.

What's the correct order to clean Airflow DAG metadata, and why not clear metadata before deleting the file?​

Delete the file first, then clean metadata. If you reverse it, the .py file still exists, so dag-processor re-registers the cleared dag row on its next scan — the metadata "comes back to life." Only by making the file vanish first (so the processor has nothing to register) and then cleaning the residual rows can you fully retire the DAG.

When an Airflow task calls ti.xcom_push() to pass pandas-processed results downstream, the task crashes outright — ValueError: Out of range float values are not JSON compliant: nan — and the app's custom logs table shows no error at all.

Encountered this while building AI Ops — an LLM-powered analytics pipeline where Airflow DAGs pull SQL data, process it with pandas, and pass results between tasks via XCom.

TL;DR​

XCom serializes with JSON under the hood, and Airflow calls json.dumps(..., allow_nan=False) to follow the JSON spec strictly — which has no NaN / Infinity whatsoever. The moment a float NaN (converted from SQL NULL by pandas) enters the data passed to xcom_push, serialization throws ValueError. Fix: recursively walk the data before push and convert NaN / ±Inf to None (JSON null).

Symptoms​

A quarterly report DAG failed, but the symptom was baffling — in the app's custom logs table, steps 1–4 for that trace were all fine, "analysis done" even logged twice (a retry), then it cut off: step 5 missing, with no error row at all:

trace=91c126c3
├─ step 1  SQL fetch            ✅
├─ step 2  pandas process       ✅
├─ step 3  rule judge           ✅
├─ step 4  LLM analysis done    ✅  ← retried after this
└─ step 5  XCom push result     ❌  ← missing, no error record

The real traceback was only in the Airflow task log:

# inside container /opt/airflow/logs/dag_id=ai_analysis_v2/run_id=.../task_id=analyze_results/attempt=N.log
ValueError: Out of range float values are not JSON compliant: nan
  File ".../ai_analysis_tasks.py", line 142, in analyze_results
    ti.xcom_push(key='sql_metadata', value=result)

The crash landed exactly on ti.xcom_push — the instant the task pushed results into XCom.

Root Cause​

Three layers stack up, all required:

1. The JSON spec has no NaN / Infinity. RFC 8259 only allows finite numeric literals. Python's json.dumps will happily emit bare NaN and Infinity by default, but those are Python-specific extensions, not valid JSON — any strict parser (Airflow included) rejects them.

2. Airflow XCom serializes with allow_nan=False. XCom's default JSON serializer explicitly disables NaN tolerance, so encountering NaN throws ValueError: Out of range float values are not JSON compliant instead of silently emitting invalid JSON.

3. pandas reads SQL NULL as NaN. pandas.read_sql returns float('nan') for SQL NULL columns. Once such a column flows through computation and to_dict('records') into the result object, NaN hitches a ride into xcom_push:

import pandas as pd

# A SQL NULL cell → pandas reads it as NaN
df = pd.DataFrame({"ad_roi": [1.2, None, 0.8]})
records = df.to_dict("records")
# [{'ad_roi': 1.2}, {'ad_roi': nan}, {'ad_roi': 0.8}]   ← nan slipped in

# downstream task crashes on push
ti.xcom_push(key="result", value=records)
# ValueError: Out of range float values are not JSON compliant: nan

This stayed latent for a long time because the data usually had values in those columns; it only surfaced when a client had zero ad spend for an entire quarter and ad_roi came back NULL across the board — the first time NaN entered the XCom path at scale.

Why no error in the logs table? Because the crash happens during XCom serialization, outside the task function's try/except — the exception bubbles straight up to the Airflow scheduler and only lands in Airflow's own task log. The app's custom logs table catch never gets a chance to record it. That's what makes this failure so confusing: it looks "silent."

Solution​

Scrub all NaN / ±Inf from the data before it enters XCom.

1. Write a pure recursive cleaner​

import math

def json_safe_value(obj):
    """
    Recursively convert NaN / +Inf / -Inf to None so the data is
    strictly JSON-serializable. Handles dict / list / tuple / scalar;
    unknown types pass through unchanged.
    """
    if isinstance(obj, float):
        if math.isnan(obj) or math.isinf(obj):
            return None
        return obj
    if isinstance(obj, dict):
        return {k: json_safe_value(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [json_safe_value(v) for v in obj]
    return obj

Why not df.fillna(None)? Because fillna(None) on numeric columns is unstable across pandas versions and dtypes — sometimes it coerces the dtype instead of nulling values. It also only handles DataFrames, not floats already nested inside dicts/lists after to_dict. Recursive cleaning at the "data is now native Python structures" layer is the most robust fallback.

2. Centralize the guard before push​

The worry-free approach is to hang the cleanup on the single chokepoint all xcom_push calls go through, rather than remembering to call it at every push site:

def push_safe(ti, key, value):
    """Clean NaN/Inf before XCom push to prevent serialization crashes."""
    ti.xcom_push(key=key, value=json_safe_value(value))

# inside the task
push_safe(ti, "sql_metadata", result)
push_safe(ti, "processor_output", processor_result)

3. Fix the "silent failure" observability gap​

Fixing serialization alone isn't enough — the gap where exceptions outside try/except never reach the app's logs table must be closed too. Attach a failure decorator that logs the top-level exception to your table before re-raising:

import functools
import logging

logger = logging.getLogger(__name__)

def log_task_failure(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except Exception:
            logger.error("task %s failed", fn.__name__, exc_info=True)
            # write the traceback into the app's custom logs table here
            raise
    return wrapper

@log_task_failure
def analyze_results(**context):
    ...

Now if another exception slips outside a catch, the app's logs table still gets an error row — no more "silent failure."

After the fix, rerunning the same conf: DAG all green, DB write success, and the formerly-NaN ad_roi lands as null in the database; downstream is happy.

On the same Airflow analytics pipeline, this isn't the only way data silently misbehaves — PostgresHook silently dropping multi-statement SQL results is another classic.

Notes​

FAQ​

How to fix Airflow "Out of range float values are not JSON compliant"?​

XCom serializes with json.dumps(allow_nan=False) and ran into NaN / Infinity, which the JSON spec does not allow. The usual root cause is pandas reading a SQL NULL into float('nan') that then flows into xcom_push. Fix it by recursively converting NaN / ±Inf to None (JSON null) before push, centralized in a pure json_safe_value helper.

Why does my Airflow task fail but my custom logs table has no error?​

If the exception happens during XCom serialization, outside the task function's try/except, it only bubbles up to the Airflow scheduler and lands in the Airflow task log (inside the container at /opt/airflow/logs/). The app's custom logs table catch never sees it, so it looks like a "silent failure." To triage, read the Airflow task log traceback directly instead of only checking app logs.

Can Airflow XCom store pandas NaN directly?​

No. XCom defaults to JSON serialization, and the JSON spec only has finite numbers — no NaN / Infinity. The right fix is to convert NaN to None (JSON null) before push. Switching to binary object serialization sidesteps the type limit but produces unreadable, non-portable values with deserialization security risk; not recommended for production.

When an Airflow DAG reads a .sql template file as a single string and passes it to PostgresHook.get_pandas_df(), prior SELECT results are silently dropped — the DAG reports "SQL query returned no results", but copying the same SQL into psql returns data normally.

Encountered this while building AI Analytics — an LLM-powered analytics pipeline where an Airflow DAG reads multi-query report templates from .sql files and executes them.

TL;DR​

PostgresHook.get_pandas_df(sql) internally calls pandas.read_sql(sql, conn) → psycopg2 cursor.execute(sql). When sql is a single string with multiple ;-separated SELECTs, the DBAPI only exposes the cursor of the last result set — prior query results are silently dropped with no error. Fix: split by top-level semicolons into a list[str] and call get_pandas_df per statement, or pass the list directly so DbApiHook runs them sequentially.

Symptom​

The DAG task executing shop_monthly_overview.sql reports "SQL query returned no results":

sql_count = 1   ← template clearly contains 4 queries
result = "❌ SQL query returned no results"

But the same SQL pasted into psql against the same database with the same parameters returns data for all 4 SELECTs.

Reproduction​

Verify get_pandas_df behavior with multi-statement SQL directly inside the Airflow container:

from airflow.providers.postgres.hooks.postgres import PostgresHook

hook = PostgresHook(postgres_conn_id="postgres_default")

# Three SELECTs concatenated into one string
sql = "SELECT 1 AS a; SELECT 2 AS b; SELECT 99 AS c WHERE 1=0;"

df = hook.get_pandas_df(sql)
print(df.columns.tolist())  # ['c']  ← only got the last statement's columns
print(df)                    # Empty  ← the last statement itself returns 0 rows

Expected three result sets, got only the last one (SELECT 99 ... WHERE 1=0, 0 rows). The first two completely disappear with no error or warning.

Root Cause​

The call chain is PostgresHook.get_pandas_df → DbApiHook.get_pandas_df → pandas.io.sql.read_sql → psycopg2 cursor.execute(sql).

The DBAPI protocol (PEP 249) allows execute to accept a string with multiple ;-separated statements. PostgreSQL executes all of them, but the cursor only exposes the last result set — this is inherent PostgreSQL wire protocol behavior, not an Airflow or pandas bug.

┌────────────────────────────────────────────────────┐
│ SELECT 1;  ← executed, result set 1 dropped at once │
│ SELECT 2;  ← executed, result set 2 dropped at once │
│ SELECT 99 WHERE 1=0;  ← executed, result set 3 exposed │
└────────────────────────────────────────────────────┘
                       ↓
            pandas.read_sql only fetches result set 3

The root cause in our code is task_execute_sql reading the entire .sql file as a single string and passing it to get_pandas_df:

# ❌ Problematic code
sql_text = open(sql_path).read()  # full text with 4 SELECTs
df = pg_hook.get_pandas_df(sql_text)  # only gets the last result

Why does psql return data? Because the psql frontend actively iterates through all result sets and prints them one by one, while a DBAPI cursor does not.

Solution​

Option A (Recommended): Split by top-level semicolon, execute one by one​

Suitable for .sql template files — they contain comments, quotes, and multiple queries that need robust splitting.

def split_sql_statements(sql: str) -> list:
    """
    Split SQL by top-level semicolons, correctly handling:
    - Semicolons inside single-quoted strings ('a;b' is not split)
    - SQL-standard '' escape ('it''s' is not split)
    - Semicolons inside -- line comments (-- note; not split is not split)
    """
    statements = []
    buf = []
    i, n = 0, len(sql)
    in_quote = False

    while i < n:
        ch = sql[i]

        # Inside a single-quoted string
        if in_quote:
            buf.append(ch)
            if ch == "'":
                # '' = literal single quote, does not end the string
                if i + 1 < n and sql[i + 1] == "'":
                    buf.append(sql[i + 1])
                    i += 2
                    continue
                in_quote = False
            i += 1
            continue

        # Top level
        if ch == "'":
            in_quote = True
            buf.append(ch)
        elif ch == '-' and i + 1 < n and sql[i + 1] == '-':
            # Line comment, swallow to end of line (; inside is not a split point)
            while i < n and sql[i] != '\n':
                buf.append(sql[i])
                i += 1
            continue
        elif ch == ';':
            stmt = ''.join(buf).strip()
            if stmt:
                statements.append(stmt)
            buf = []
            i += 1
            continue
        else:
            buf.append(ch)
        i += 1

    # Trailing block without a final semicolon
    stmt = ''.join(buf).strip()
    if stmt:
        statements.append(stmt)

    return statements


# Caller
sql_text = open(sql_path).read()
statements = split_sql_statements(sql_text)

# Execute one by one, collect all results
all_results = []
for idx, stmt in enumerate(statements, start=1):
    df = pg_hook.get_pandas_df(stmt)
    if not df.empty:
        all_results.append({
            "sql_index": idx,
            "sql": stmt,
            "data": df.to_dict("records"),
            "columns": df.columns.tolist(),
            "row_count": len(df),
        })

Option B: Pass a list directly to DbApiHook​

Airflow's DbApiHook.run and get_records accept list[str] and execute sequentially — but get_pandas_df return behavior in list mode is inconsistent across providers. For production, Option A gives you full control.

Why not sqlparse.split?​

Community answers often recommend sqlparse.split(sqlparse.format(sql, strip_comments=True)), but strip_comments=True discards comments. If your downstream processor depends on metadata in comments (e.g. -- dimension: shop), you lose context. A hand-rolled splitter preserves the original comment text and gives you control.

Caveats​

FAQ​

How do I execute multiple SQL statements in Airflow PostgresHook?​

Pass list[str] instead of a single string. DbApiHook.get_pandas_df and run accept sql as a list and execute sequentially; a single string with semicolon-separated statements causes psycopg2 to return only the last result set. For production, split yourself and call per-statement so you control result aggregation and sql_index.

Why does get_pandas_df only return the last result for multi-statement SQL?​

pandas.io.sql.read_sql calls psycopg2 cursor.execute with the full string; the DBAPI protocol only exposes the cursor of the last result set for multi-statement execution, and prior SELECT results are dropped by the server immediately, with no error or warning. psql returns data because its frontend actively iterates all result sets, while a DBAPI cursor does not.

How do I split SQL by semicolon safely with comments and quotes?​

Scan character by character and split only at top-level semicolons outside single-quoted strings and -- line comments. Single-quote literals use the SQL-standard '' escape; do not use str.split(';'), it will mis-cut semicolons inside comments and strings. If you use sqlparse.split, note that strip_comments=True discards the original comment text.

While maintaining a 2-core 7GB cloud server for a client, I discovered disk usage at 93% (only 3GB free) and Airflow scheduler consistently consuming 160% CPU. Here's the complete diagnosis and fix.