Airflow算子能否在外部服务运行并同步DAG进度？外部服务如何上报任务状态

Airflow完全支持让外部服务告知任务启动/完成状态、无需在本地运行算子的场景，以下是几种可行的实现方式：

1. 调用Airflow REST API（推荐方案） #

Airflow 2.x及以上版本提供原生REST API，外部服务可直接通过API更新指定任务实例的状态，无需依赖Airflow本地调度执行算子。

关键步骤： #

• 确保目标DAG的运行实例（DAG Run）已创建：可通过Airflow UI触发、API调用或CLI命令生成对应的DAG Run，每个DAG Run对应唯一的dag_run_id或execution_date。
• 调用状态更新接口：使用POST /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/set_state端点，传入目标状态即可。

示例请求（curl）： #

curl -X POST \
  http://<你的Airflow WebServer地址>/api/v1/dags/your_dag_id/dagRuns/202405200000/taskInstances/bash_task1/set_state \
  -H "Authorization: Bearer <你的认证Token>" \
  -H "Content-Type: application/json" \
  -d '{"state": "running"}'  # 启动状态用running，完成用success，失败用failed

支持的状态值包括：running、success、failed、skipped，完全覆盖状态更新需求。

2. 自定义轻量Operator #

如果不想直接调用API，可以自定义一个仅负责状态同步的Operator，替代原有BashOperator。这个Operator不执行业务逻辑，仅与外部服务交互获取状态并同步到Airflow。

示例代码： #

from airflow.models.baseoperator import BaseOperator
from airflow.utils.state import State
import requests

class ExternalSyncOperator(BaseOperator):
    def __init__(self, external_service_url, external_task_id, **kwargs):
        self.external_service_url = external_service_url
        self.external_task_id = external_task_id
        super().__init__(**kwargs)

    def execute(self, context):
        # 调用外部服务接口获取任务状态
        resp = requests.get(f"{self.external_service_url}/tasks/{self.external_task_id}/status")
        external_status = resp.json().get("status")
        
        # 同步状态到Airflow任务实例
        task_instance = context["task_instance"]
        if external_status == "started":
            task_instance.state = State.RUNNING
        elif external_status == "completed":
            task_instance.state = State.SUCCESS
        elif external_status == "failed":
            task_instance.state = State.FAILED
        task_instance.update()

使用方式： #

将原有DAG中的BashOperator替换为这个自定义Operator即可：

from airflow import DAG
from your_module import ExternalSyncOperator

with DAG(dag_id="your_dag_id", ...) as dag:
    t1 = ExternalSyncOperator(
        task_id="bash_task1",
        external_service_url="http://你的外部服务地址",
        external_task_id="ext_task_1"
    )
    t2 = ExternalSyncOperator(
        task_id="bash_task2",
        external_service_url="http://你的外部服务地址",
        external_task_id="ext_task_2"
    )
    t1 >> t2

3. 使用Airflow CLI命令 #

外部服务也可通过调用Airflow CLI命令更新任务状态，适合无法使用REST API的场景：

示例命令： #

# 设置bash_task1为成功状态，参数依次为DAG ID、任务ID、DAG Run的execution_date/ID、目标状态
airflow tasks set-state your_dag_id bash_task1 2024-05-20T00:00:00+00:00 success

注意事项 #

• 无论使用哪种方式，都需要确保对应的DAG Run和Task Instance已存在，Airflow才能找到目标任务进行状态更新。
• 如果需要完全脱离Airflow调度仅用其展示状态，可先通过API创建DAG Run（POST /api/v1/dags/{dag_id}/dagRuns），再更新任务状态。

内容的提问来源于stack exchange，提问作者Phillip Ng