时间戳

首页 > 解决方案 > 气流 DAG 'start_date' 未注册

airflow - 气流 DAG 'start_date' 未注册

问题描述

我有一个正确指向我正在使用的 DAGS 的 Airflow 环境设置。一个 DAG 未加载，因为其导入失败

[2020-06-05 19:09:15,814] {dagbag.py:205} ERROR - Failed to import: /home/…/tmp.py
Traceback (most recent call last):
...
  File "/data/environments/airflow/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 511, in dag
    dag.add_task(self)
  File "/data/environments/airflow/lib/python3.7/site-packages/airflow/models/dag.py", line 1124, in add_task
    raise AirflowException("Task is missing the start_date parameter")
airflow.exceptions.AirflowException: Task is missing the start_date parameter

异常一直说我的开始日期丢失，但这不可能是正确的，因为这是 DAG 定义

dag_args = {
    'owner': 'owner',
    'depends_on_past': False,
    'start_date': datetime(2020, 1, 1),
    'email': [email],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    dag_id=dag_name,
    default_args=dag_args,
    schedule_interval=daily_at_12_UTC
)

我已经关注了文档，对此没有任何其他内容。是格式/语法问题吗？

标签： airflowdirected-acyclic-graphs

解决方案

由于 OP 从未跟进，我将在这里留下我的评论作为下面的答案。注意倒数第二行是键（即'dag=dag'）。

dag_args = {
    'owner': 'owner',
    'depends_on_past': False,
    'start_date': datetime(2020, 1, 1),
    'email': [email],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    dag_id=dag_name,
    default_args=dag_args,
    schedule_interval=daily_at_12_UTC,
    dag=dag
)

推荐阅读