Airflow DAG触发JSONDecodeError报错，求助排查解决

提供的DAG代码 #

import airflow
from datetime import timedelta, datetime
from airflow import DAG
from airflow.models import Variable
from airflow.contrib.operators.bigquery_operator import BigQueryOperator

default_args = {
"owner": "scheduled_queries",
"depends_on_past": False,
"start_date": airflow.utils.dates.days_ago(1),
"email": Variable.get("email"),
"email_on_failure": True,
"email_on_retry": True,
"retries": 0,
"retry_delay": timedelta(minutes=1),
}

dag = DAG('scheduled_queries',
schedule_interval=None, #'0 0 * * *',
#template_searchpath=tmpl_search_path,
default_args=default_args
)

报错信息 #

Broken DAG (dags/testdag.py):
Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/opt/python3.8/lib/python3.8/json/decoder.py", line 353, in raw_decode
    obj, end = self.scan_once(s, idx)
json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 1 (char 263)

问题原因 #

报错指向JSON解析失败，结合DAG代码和“其他环境正常”的前提，问题大概率出在Airflow变量email的内容上：

• 当前环境中email变量存储的内容不符合JSON规范，比如使用了单引号而非双引号，或者存在语法错误（比如多邮箱格式错误、多余的逗号等）
• Airflow的Variable.get()默认会尝试将变量值解析为JSON（deserialize_json=True是默认参数），如果变量值不是合法JSON就会触发这个报错

解决方案 #

1. 检查并修正email变量内容
   • 进入Airflow UI的Admin > Variables页面，找到email变量
   • 如果是单个邮箱，直接填写字符串（不需要引号，比如user@example.com），或者用合法JSON格式的字符串（"user@example.com"）
   • 如果是多个邮箱，必须用JSON数组格式，比如["user1@example.com", "user2@example.com"]，注意使用双引号，逗号分隔无多余符号
2. 强制关闭JSON解析（可选）
   如果不需要将email变量解析为JSON，可以修改代码中的Variable.get调用，指定deserialize_json=False：

   "email": Variable.get("email", deserialize_json=False),
   

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