2 篇博文 含有标签「JSON」

在用 json.dumps(data, default=str) 把一个含 Python set 的字典持久化、再回读用 in 判断成员时，结果静默出错——没有任何报错，但 in 判断全乱。

在开发 AI运营 时遇到此问题——基于大语言模型的智能分析，自动洞察市场趋势、用户行为、销售数据，提供精准运营策略。某次分析模板的决策重放（replay）功能里，需要把「缺失月份集合」序列化进快照、回放时再读出来判断某月是否缺失。结果重放后，本应判定为「缺失」的月份被误判为「不缺失」，而整个链路没有任何异常抛出。

TL;DR​

default=str 不是万能兜底。它会把 set 交给 str()，在 JSON 里存成 "{1, 2}" 这样的字面量字符串而非数组；回读后类型已不可逆，对它做 in 判断会退化成子串匹配，静默返回错误结果。涉及 set 时，正确做法是序列化前转 list、读取时 set() 重建。

问题现象​

下面这段代码完整复现了静默出错的过程：

import json

# 一个含 set 的字典——比如"需要补数据的缺失月份"
data = {"missing_months": {"3", "5", "12"}}

# 用 default=str 兜底序列化（常见的"别让它报错"写法）
serialized = json.dumps(data, default=str)
print(serialized)
# {"missing_months": "{'3', '5', '12'}"}   ← 变成了字符串，不是数组！

# 回读
back = json.loads(serialized)
value = back["missing_months"]
print(type(value))   # <class 'str'>   ← 已经不是 set 了

# 静默 bug：本想判断某月份是否在"缺失集合"里
print("1" in value)   # True   ← 1 根本不在 {3,5,12}，但 "1" 是 "12" 的子串！
print("3" in value)   # True   ← 碰巧对
print("9" in value)   # False

"1" in value 返回 True，但原集合 {"3", "5", "12"} 根本不含 "1"。没有异常、没有警告，判断结果就这样悄悄错了。这种 bug 在依赖判断结果做分支（如「这个月缺数据吗？缺则补采」）的链路里尤其致命。

根因​

分三层看：

第一层：set 本就不可 JSON 序列化。 JSON 只有 array（对应 list）和 object，没有集合类型。直接 json.dumps({"x": {1, 2}}) 会抛 TypeError: Object of type set is not JSON serializable。

第二层：default=str 把报错变成了静默污染。 json.dumps 的 default 参数在遇到无法序列化的对象时被调用，期望返回一个可序列化的值。str 作为 default 时，会把对象交给 str()——set 就被转成了它的 Python 字面量表示 {'3', '5', '12'}，作为字符串存进 JSON：

>>> json.dumps({"m": {"3", "5", "12"}}, default=str)
'{"m": "{\'3\', \'5\', \'12\'}"}'

报错消失了，代价是类型从 set 变成了 str，且这个过程不会给你任何提示。

第三层：in 对 str 和 set 语义不同。 这是静默 bug 的核心。对 set/list，x in s 是成员判断；对 str，x in s 退化成子串匹配。回读后的值是字符串 "{'3', '5', '12'}"，于是 "1" in "{'3', '5', '12'}" 判断的是字符 "1" 是否作为子串出现——而 "12" 里恰好有 "1"，所以返回 True。

这和 Airflow PostgresHook 多语句 SQL 静默丢结果 是同一类陷阱：最危险的 bug 不是抛异常，而是「静默地给错结果」，因为没有任何信号提醒你去查。

解决方案​

核心原则：JSON 里只存标准类型，集合语义在读取端重建。

方案一：序列化前显式转 list（推荐）​

最直接、最可控——明确知道哪里有 set，就地转成 list：

import json

# 序列化前：set → list（标准 JSON 数组）
data = {"missing_months": list({"3", "5", "12"})}
serialized = json.dumps(data)
print(serialized)
# {"missing_months": ["3", "5", "12"]}   ← 正确的 JSON 数组

# 回读后重建 set
back = json.loads(serialized)
months = set(back["missing_months"])
print("1" in months)   # False ✓
print("3" in months)   # True  ✓

序列化结果是一个干净的 JSON 数组，跨语言、可读、可还原。

方案二：自定义 default 函数（数据来源复杂时）​

如果数据结构较深、不确定哪里混入了 set，用一个专门处理集合类型的 default 函数，既不丢失语义，又能兜底其他非标准类型：

import json

def safe_default(obj):
    # 集合类型 → list，保留为标准 JSON 数组
    if isinstance(obj, (set, frozenset)):
        return sorted(obj)          # 排序让输出稳定可预测
    # 其他无法序列化的类型再退回 str，但要清楚这会丢类型
    return str(obj)

data = {"missing_months": {"3", "5", "12"}, "created_at": some_datetime}
serialized = json.dumps(data, default=safe_default)
# {"missing_months": ["3", "5", "12"], "created_at": "..."}

back = json.loads(serialized)
months = set(back["missing_months"])
print("1" in months)   # False ✓

相比无脑 default=str，这个函数把「需要保真的类型」（集合）单独处理，只有真正无法表示的类型才退回 str，把静默风险控制到最小。

常见问题​

Python set 怎么转 json？​

set 不是 JSON 原生类型，直接 json.dumps 会抛 TypeError。正确做法是序列化前用 list(set) 转成列表，存成标准 JSON 数组；读取时再 set(back["key"]) 重建。这样既不报错，又能完整还原集合语义，跨语言也兼容。

json.dumps 报 Object of type set is not JSON serializable 怎么解决？​

根因是 set 不可 JSON 序列化。最稳妥的解法是序列化前把 set 转成 list；也可以传一个 default 函数，在里面对 isinstance(obj, (set, frozenset)) 返回 list(obj)。要避免用 default=str 兜底——它虽不报错，却把 set 存成了字符串，回读后类型无法还原。

为什么 default=str 序列化 set 后 in 判断结果错了？​

default=str 会把 set 交给 str()，变成字面量字符串 '{1, 2}' 存进 JSON。回读后值类型是 str 而非 set，x in s 就从「成员判断」退化成「子串匹配」——比如 "1" in "{'3','5','12'}" 因 "12" 含字符 "1" 而返回 True，但原集合并不含 "1"。解法是序列化 list、读取时 set() 重建。

在 Airflow 任务用 ti.xcom_push() 把 pandas 处理后的结果推给下游任务时，任务直接崩溃——ValueError: Out of range float values are not JSON compliant: nan，而且应用自建的 logs 表里找不到任何错误记录。

在开发 AI运营 时遇到此问题——基于大语言模型的智能分析流水线，Airflow DAG 从 SQL 取数、经 pandas 处理后，通过 XCom 在任务间传递结果。

TL;DR​

XCom 底层用 JSON 序列化，而 Airflow 调用 json.dumps(..., allow_nan=False) 严格遵循 JSON 标准——标准里压根没有 NaN / Infinity。pandas 从 SQL NULL 转来的浮点 NaN 一旦进了 xcom_push 的数据，序列化立即抛 ValueError。解法：在 push 前递归遍历数据，把 NaN / ±Inf 转成 None（JSON null）。

问题现象​

某次季度报表 DAG 失败，但症状很迷惑——应用自建的 logs 表里，该 trace 的步骤 1–4 全部正常，连「分析完成」都打了两次（重试），然后就断了，步骤 5 缺失，且没有任何 error 行：

trace=91c126c3
├─ step 1  SQL 取数            ✅
├─ step 2  pandas 处理         ✅
├─ step 3  规则判定            ✅
├─ step 4  LLM 分析完成        ✅  ← 之后重试了一次
└─ step 5  XCom 推送结果       ❌  ← 缺失，无 error 记录

去 Airflow 任务日志才看到真正的 traceback：

# 容器内 /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)

崩溃点精确落在 ti.xcom_push——任务把结果推给 XCom 的那一刻。

根因​

三层叠加，缺一不可：

1. JSON 标准不含 NaN / Infinity。 RFC 8259 定义的 JSON 只允许数字字面量是有限数。虽然 Python 的 json.dumps 默认会把 NaN 写成裸 NaN、把 Infinity 写成 Infinity，但这是 Python 的私有扩展，不是合法 JSON——任何严格 parser（包括 Airflow 用的）读到都会拒绝。

2. Airflow XCom 序列化时显式 allow_nan=False。 XCom 默认走 JSON serializer，序列化时关闭了 NaN 容忍，遇到 NaN 直接抛 ValueError: Out of range float values are not JSON compliant，而不是偷偷写出非法 JSON。

3. pandas 把 SQL NULL 读成 NaN。 pandas.read_sql 对 SQL 的 NULL 列返回 float('nan')。一旦这列参与计算后被 to_dict('records') 带进结果对象，NaN 就顺着数据流进了 xcom_push：

import pandas as pd

# SQL 某行某列是 NULL → pandas 读成 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 混了进来

# 下游任务 push 时崩
ti.xcom_push(key="result", value=records)
# ValueError: Out of range float values are not JSON compliant: nan

这次之所以长期没触发，是因为平时跑的数据那些列都有值；直到某客户某个季度完全没有广告投放、ad_roi 整列 NULL，NaN 才第一次大规模进入 XCom 路径。

为什么 logs 表没有 error？ 因为崩溃发生在 XCom 序列化阶段，处于任务函数的 try/except 之外——异常直接冒泡给 Airflow 调度器，只写进 Airflow 自己的任务日志，应用层自建的 logs 表的 catch 根本没机会记录。这是这类故障最迷惑的地方：看起来「无声失败」。

解决方案​

在数据进入 XCom 前，递归清洗掉所有 NaN / ±Inf。

1. 写一个纯函数递归清洗​

import math

def json_safe_value(obj):
    """
    递归把 NaN / +Inf / -Inf 转成 None，使数据可被 JSON 严格序列化。
    兼容 dict / list / tuple / scalar，遇到未知类型原样返回。
    """
    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

为什么不能用 df.fillna(None)？因为 pandas 的 fillna(None) 在数值列上行为依版本和 dtype 不稳定，有时会把 NaN 强制转型而非置空；而且它只处理 DataFrame，管不到已经 to_dict 之后嵌在 dict/list 里的浮点。递归清洗在「数据已变成 Python 原生结构」这一层兜底，最稳。

2. 在 push 前统一兜底​

最省心的做法是把清洗挂在所有 xcom_push 的必经之路上（比如一个归一化函数），而不是每个 push 点都记得调：

def push_safe(ti, key, value):
    """XCom push 前清洗 NaN/Inf，杜绝序列化崩溃。"""
    ti.xcom_push(key=key, value=json_safe_value(value))

# 任务内
push_safe(ti, "sql_metadata", result)
push_safe(ti, "processor_output", processor_result)

3. 补上「无声失败」的可观测性​

光修序列化还不够——异常发生在 catch 外、应用 logs 表不记录这个缺口要一起补。给任务挂一个失败装饰器，顶层异常先落库再 re-raise：

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)
            # 这里把 traceback 写进应用自建 logs 表
            raise
    return wrapper

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

这样即使以后再出现 catch 外的异常，应用 logs 表也能留下 error 行，不再「无声失败」。

修完后重跑同一份 conf：DAG 全绿、落库 success，原本 NaN 的 ad_roi 在库里落成 null，下游正常。

同一条 Airflow 分析流水线上，让数据悄悄出问题的坑不止这一个——PostgresHook 多语句 SQL 静默丢结果是另一个典型案例。

注意事项​

常见问题​

Airflow 报 Out of range float values are not JSON compliant 怎么解决？​

这是 XCom 用 json.dumps(allow_nan=False) 序列化时遇到了 NaN / Infinity，而 JSON 标准不含这两种值。根因通常是 pandas 把 SQL NULL 读成了 float('nan')，跟着数据流进了 xcom_push。解法是在 push 前递归把 NaN / ±Inf 转成 None（JSON null），用一个 json_safe_value 纯函数统一兜底即可。

为什么 Airflow 任务失败但自建 logs 表没有错误记录？​

如果异常发生在 XCom 序列化阶段、且位于任务函数的 try/except 之外，错误只会冒泡给 Airflow 调度器、写进 Airflow 任务日志（容器内 /opt/airflow/logs/），应用层自建的 logs 表的 catch 拿不到，于是表现为「无声失败」。排查这类情况要直接看 Airflow task log 的 traceback，别只盯应用日志。

Airflow XCom 能不能直接存 pandas 的 NaN？​

不能。XCom 默认走 JSON 序列化，而 JSON 标准只有有限数字，没有 NaN / Infinity。正确做法是 push 前把 NaN 转成 None（对应 JSON null）。换成二进制对象序列化虽能绕过类型限制，但结果不可读、不跨进程/版本、还有反序列化安全风险，生产环境不推荐。