Airflow中使用Python向BigQuery加载JSON数据时XCom序列化错误如何解决

你遇到的两个报错均和Airflow XCom的默认序列化逻辑有关，和BigQuery写入逻辑本身无关：

• Airflow PythonOperator默认会将执行函数的返回值自动存入XCom，用于下游任务拉取使用
• Airflow默认使用JSON格式序列化XCom值，仅支持字符串、数字、列表、字典等基础类型
• 你代码中返回的job.result()是BigQuery SDK定义的LoadJob自定义对象，没有原生JSON序列化实现，因此触发序列化错误

按推荐优先级排序如下：

方案1：修改返回值（最推荐，生产环境可用） #

不需要返回完整LoadJob对象的前提下，仅返回可序列化的基础类型，或者直接不返回任何值：

• 如果不需要下游任务使用该任务的输出，直接删除return job.result()语句即可，没有返回值就不会触发XCom写入逻辑
• 如果需要返回执行结果做后续判断，仅提取基础类型的统计信息返回：

# 替换原return语句
load_result = job.result()
return {
    "job_state": load_result.state,
    "loaded_rows": load_result.output_rows,
    "error_info": load_result.errors
}

方案2：开启XCom Pickle支持（不推荐生产环境使用） #

如果确实需要传递完整LoadJob对象给下游任务，可以修改Airflow配置开启Pickle序列化：

1. 打开Airflow配置文件airflow.cfg
2. 找到[core]配置段，修改参数为enable_xcom_pickling = True
3. 重启Airflow调度器与Worker节点生效

注意：Pickle序列化存在安全风险，若XCom中存入恶意构造的Pickle对象，反序列化时会执行任意代码，生产环境需谨慎使用。

额外优化提示 #

你当前代码存在一处遗漏导入：使用了json.load()但未导入json模块，需在代码开头添加import json，否则会触发NameError报错。
另外建议不要硬编码Google服务密钥路径，可通过Airflow环境变量或GCP连接管理来配置敏感信息，降低配置泄露风险。

内容的提问来源于stack exchange，提问作者Shadow Walker