0

每当我尝试运行我的 DAG 时，它都会获得成功状态，但它无法运行该任务。第一次运行时它完美运行

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 时，我收到以下错误， 依赖项“任务实例状态”失败：任务处于“成功”状态，这不是执行的有效状态。必须清除任务才能运行

4

2 回答 2

0

而不是schedule_interval='@once'尝试schedule_interval=None。

@once意味着它只会运行一次。

于 2019-05-30T15:19:06.193 回答
0

您start_date在两个 DAGS 中的变量都是动态的，可能会导致您的问题'start_date': datetime.utcnow()。不建议将它们设置为动态启动，这会导致错误。

尝试将其设置为静态开始日期，例如'start_date': datetime(2019, 5, 29) #year month day

取自Airflow FAQ

我们建议不要将动态值用作 start_date，尤其是 datetime.now()，因为它可能会造成混淆。一旦周期结束，任务就会被触发，理论上，@hourly DAG 永远不会到达现在之后的一个小时，因为 now() 会继续进行。

另一个关于此的问题：为什么动态启动会导致问题

于 2019-05-30T16:04:17.843 回答