Не удается запустить задачу в Airflow

Question

Всякий раз, когда я пытался запустить свою группу обеспечения доступности баз данных, он получает состояние успеха, но не может выполнить задачу. В первый раз работает отлично

trigger_controller_dag.py:

def conditionally_trigger(context, dag_run_obj):
    c_p = context['params']['condition_param']
    if context['params']['condition_param']:
        dag_run_obj.payload = {'message': context['params']['message']}
        pp.pprint(dag_run_obj.payload)
        return dag_run_obj

dag = DAG(
    dag_id='example_trigger_controller_dag',
    default_args={
        "owner": "airflow",
        "start_date": datetime.utcnow(),
    },
    schedule_interval='@once',
)

trigger = TriggerDagRunOperator(
    task_id='test_trigger_dagrun',
    trigger_dag_id="example_trigger_target_dag",
    python_callable=conditionally_trigger,
    params={'condition_param': True, 'message': 'Hello World'},
    dag=dag,
)

trigger_target_dag.py:

args = {
    'start_date': datetime.utcnow(),
    'owner': 'airflow',
}

dag = DAG(
    dag_id='example_trigger_target_dag',
    default_args=args,
    schedule_interval=None,
)


def run_this_func(ds, **kwargs):
    print("Remotely received value of {} for key=message".
          format(kwargs['dag_run'].conf['message']))


run_this = PythonOperator(
    task_id='run_this',
    provide_context=True,
    python_callable=run_this_func,
    dag=dag,
)

Во время работы DAG появляется следующая ошибка: зависимость 'Состояние экземпляра задачи' СБОЙ: Задача находится в состоянии «успех», которое не является допустимым для выполнения. Задача должна быть очищена для запуска

Answer 1

Ваши start_date переменные в обеих DAGS являются динамическими и могут вызывать вашу проблему 'start_date': datetime.utcnow(). Установка их в динамический запуск не рекомендуется и приводит к ошибкам.

Попробуйте установить статическую дату начала, например 'start_date': datetime(2019, 5, 29) #year month day

Взято из FAQ по воздушному потоку

Мы рекомендуем не использовать динамические значения в качестве start_date, особенно datetime.now (), поскольку это может быть довольно запутанным. Задача запущена как только период закроется, и в теории @hourly DAG никогда не получит через час, как сейчас () движется вперед.

Еще один SO вопрос: почему проблемы с динамическим запуском

Answer 2

Вместо schedule_interval='@once' try schedule_interval=None.

@once означает, что он будет работать только один раз.