Airflow - 将给定 dag_id 和 run_id 的特定 task_id 标记为成功或失败

Question

我可以在外部(使用 http 请求？)将与 dag_id 和 run_id 关联的特定 task_id 标记为成功/失败。

我的任务是在外部系统上长时间运行的任务，我不希望我的任务轮询系统以查找状态..因为我们可能同时运行几个 1000 个任务..

理想情况下希望我的任务

发出 http 请求以开始我的外部工作

去 sleep

作业完成后，它(外部系统或我的作业的后期构建操作)通知 Airflow 任务已完成(由 task_id、dag_id 和 run_id 标识)

谢谢

Answer 1

您可以通过将 SQL 查询直接发送到 Airflow 的元数据数据库来解决此问题:

UPDATE task_instance
SET state = 'success',
    try_number = 0
WHERE
    task_id = 'YOUR-TASK-ID'
AND
    dag_id = 'YOUR-DAG-ID'
AND
    execution_date = '2019-06-27T16:56:17.789842+00:00';

备注 :

execution_date过滤器至关重要，Airflow 通过 execution_date 识别 DagRuns ，不是真的被他们run_id .这意味着您确实需要获取 DagRun 的执行/运行日期才能使其工作。

try_number = 0添加部分是因为有时 Airflow 会将任务重置回 failed如果它注意到 try_number 已经在它的极限 ( max_tries )

你可以在 Airflow 的源代码中看到它: https://github.com/apache/airflow/blob/750cb7a1a08a71b63af4ea787ae29a99cfe0a8d9/airflow/models/dagrun.py#L203