2 posts tagged with "pandas"

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.