Airflow BigQuery到GCS备份DAG无报错但任务全跳过

AI 小北

2026-9-2

故障根因

所有任务被跳过的核心原因是环境变量读取值的类型不匹配，导致日期前置校验逻辑永远不满足：

通过os.environ读取的环境变量值默认是字符串类型，你配置的BACKUP_SCHEDULE_DAY=3读取后实际为字符串"3"
datetime.isoweekday()方法返回的是整数类型的周标识（周三对应整数3），Python中整数3和字符串"3"做相等判断永远返回False
因此_check_valid_day任务每次执行都会抛出AirflowSkipException被标记为跳过，而你给所有下游备份任务设置的trigger_rule="all_success"要求上游任务必须成功才能执行，上游跳过时所有下游任务会连带被跳过，全程不会抛出业务错误。

另外你的代码还存在一个隐性故障点：DAG顶层定义的today = datetime.today()是Airflow调度器解析DAG文件时生成的固定时间，不是任务实际运行的时间，后续遇到跨天调度、补跑历史任务的场景时，会出现备份路径日期错误、校验逻辑不生效的问题。

修复方案

修复类型不匹配问题，读取备份日配置时显式转为整数：

# 原代码 schedule_day =os.environ["BACKUP_SCHEDULE_DAY"]
schedule_day = int(os.environ["BACKUP_SCHEDULE_DAY"])

替换所有硬编码的当前时间获取逻辑，改用Airflow运行时上下文的日期，避免解析时间和运行时间不一致的问题：
- 调整日期校验函数，从任务上下文取实际执行日期，不要直接调用datetime.today()：

def _check_valid_day(**kwargs):
    execution_date = kwargs["execution_date"]
    if execution_date.isoweekday() == kwargs["schedule_day"]:
        return True
    else:
        raise AirflowSkipException("does not correspond to the backup day")

备份目标路径使用Airflow内置Jinja模板渲染运行时日期，替换顶层定义的today变量拼接逻辑（注意f-string内模板大括号需要写双大括号转义）：

task_dict[task_name] = BigQueryToGCSOperator(
    task_id=task_name,
    trigger_rule="all_success",
    dag=dag,
    source_project_dataset_table=table_to_backup,
    destination_cloud_storage_uris=[
        f"gs://{destination_bucket}/{dataset}/{table}/{{ execution_date.year }}/{{ execution_date.month }}/{{ execution_date.day }}/{table}-*.avro"
    ],
    export_format="AVRO",
    compression="NONE",
    labels=None,
)

验证步骤

部署修复后的代码，等待调度器完成DAG解析（通常30秒内）
临时将BACKUP_SCHEDULE_DAY改为触发当天对应的周几数值，手动触发DAG运行
确认check_valid_day任务执行成功无跳过，后续备份任务正常启动，GCS目标路径下生成对应备份文件
验证完成后将BACKUP_SCHEDULE_DAY改回计划的备份日配置即可

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