Task is missing the start_date parameter

user_8523137 0
Viewed 2311

I have a simple dag: (3 python operations)

default_args = {
    'owner': 'default_user',
    'start_date': days_ago(2),
    'retries': 5,
    'retry_delay': timedelta(minutes=1),
}

dag = DAG(
    'basic_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

default_args
def alg_1():
    print('\nHello alg1')
    print('\nHFinshed alg1')

def alg_2():
    print('\nHello alg2')
    print('\nHFinshed alg2')

def alg_3():
    print('\nHello alg3')
    print('\nHFinshed alg3')


with DAG('my_test', description='my_test description', schedule_interval='@once', catchup=False) as dag:

    python_task_1 = PythonOperator(task_id='my_task_1', python_callable=alg_1, dag=dag)
    python_task_2 = PythonOperator(task_id='my_task_2', python_callable=alg_2, dag=dag)
    python_task_3 = PythonOperator(task_id='my_task_3', python_callable=alg_3, dag=dag)

    python_task_1 >> python_task_2
    python_task_2 >> python_task_3

I'm getting the following error (after run airflow initdb):

airflow.exceptions.AirflowException: Task is missing the start_date parameter

I don't understand why it missing, I give it in the default_args

How can I fix it ?

airflow
1 Answers

You have defined your DAG twice, and the tasks are assigned to the second DAG that does not have default_args. Use the following fixed DAG:

default_args = {
    'owner': 'default_user',
    'start_date': days_ago(2),
    'retries': 5,
    'retry_delay': timedelta(minutes=1),
}

def alg_1():
    print('\nHello alg1')
    print('\nHFinshed alg1')

def alg_2():
    print('\nHello alg2')
    print('\nHFinshed alg2')

def alg_3():
    print('\nHello alg3')
    print('\nHFinshed alg3')


with DAG('my_test', description='my_test description', catchup=False, default_args=default_args, schedule_interval=timedelta(days=1)) as dag:

    python_task_1 = PythonOperator(task_id='my_task_1', python_callable=alg_1)
    python_task_2 = PythonOperator(task_id='my_task_2', python_callable=alg_2)
    python_task_3 = PythonOperator(task_id='my_task_3', python_callable=alg_3)

    python_task_1 >> python_task_2
    python_task_2 >> python_task_3
user_56915250
Related