Airflow创建自定义DAG提示dag_id不存在无法识别问题求助

• 空白Python文件不会被Airflow识别。Airflow扫描dags目录时仅会提取文件内实例化的DAG类对象，没有合法DAG定义的文件会被直接忽略，不会进入DAG列表，自然也查询不到对应的dag_id。
• airflow dags backfill的入参是DAG定义里的dag_id参数值，不是Python文件名，就算文件内有DAG定义，如果你传的文件名和dag_id不一致也会报错。
• 如果DAG定义正确但还是看不到，大概率是Airflow Scheduler未启动、Scheduler对dags目录没有读取权限，或是还没到默认的30秒扫描间隔。
• 若DAG定义文件存在语法错误、依赖缺失，也会被Airflow跳过解析，不会出现在DAG列表，可查看scheduler日志排查具体报错。

步骤1：确认目录与权限 #

首先确认dags目录和你创建的py文件对Airflow运行用户可读，执行以下命令修正权限：

chmod 755 /home/user@local/airflow/dags
chmod 644 /home/user@local/airflow/dags/sample_dag.py

步骤2：编写合法DAG定义 #

你可以直接复制以下最小可用DAG代码到sample_dag.py中：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

# 这里的sample_dag就是dag_id，后续命令要传这个值
with DAG(
    dag_id="sample_dag",
    default_args={
        "retries": 1,
        "retry_delay": timedelta(minutes=3),
    },
    description="我的第一个测试DAG",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["test"],
) as dag:
    # 定义一个简单的测试任务
    test_task = BashOperator(
        task_id="test_print",
        bash_command="echo 'Hello Airflow!'",
    )

步骤3：验证DAG是否可被正常解析 #

执行以下命令手动触发DAG解析，确认你的DAG已被识别：

# 列出所有已识别的DAG，找sample_dag这一行
airflow dags list
# 也可以直接测试DAG运行，不会实际写入调度记录
airflow dags test sample_dag 2024-01-01

如果列表里出现了dag_id为sample_dag的记录，说明解析成功。

步骤4：启动服务并在界面查看 #

确保同时启动webserver和scheduler两个服务：

# 后台启动scheduler
airflow scheduler -D
# 后台启动webserver
airflow webserver -D

最多等待30秒，刷新localhost:8080的Web界面，就可以在DAG列表里找到sample_dag了。

步骤5：执行backfill操作 #

确认DAG存在后再执行backfill命令，注意参数是dag_id不是文件名：

airflow dags backfill sample_dag --start-date 2024-01-01 --end-date 2024-01-03

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