Airflow

Question

我目前使用的是 Airflow Taskflow API 2.0。我在结合使用 TaskGroup 和 BranchPythonOperator 时遇到问题。

下面是我的代码：

import airflow
from airflow.models import DAG
from airflow.decorators import task, dag
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import BranchPythonOperator, PythonOperator
from airflow.operators.python import task, get_current_context
from random import randint
from airflow.utils.task_group import TaskGroup


default_args = {
    'owner': 'Airflow',
    'start_date': airflow.utils.dates.days_ago(2),
}

@task
def dummy_task():
    return {}


@task
def task_b():
    return {}

@task
def task_c():
    return {}

def final_step():
    return {}

def get_tasks(**kwargs):
    task = 'task_a'

    return task


with DAG(dag_id='branch_dag', 
    default_args=default_args, 
    schedule_interval=None) as dag:

    with TaskGroup('task_a') as task_a:
        obj = dummy_task()

    tasks = BranchPythonOperator(
        task_id='check_api',
        python_callable=get_tasks,
        provide_context=True
    )

    final_step = PythonOperator(
        task_id='final_step',
        python_callable=final_step,
        trigger_rule='one_success'
    )

    b = task_b()
    c = task_c()

    tasks >> task_a >> final_step
    tasks >> b >> final_step
    tasks >> c >> final_step

当我触发此 DAG 时，我在 check_api 任务中收到以下错误：

airflow.exceptions.TaskNotFound: 找不到任务 task_a

是否可以将 TaskGroup 与 BranchPythonOperator 结合使用？

谢谢，

Answer 1

BranchPythonOperator 预计 return task_ids

您需要将 get_tasks 函数更改为：

def get_tasks(**kwargs):
    task = 'task_a.dummy_task'
    return task

Airflow - 在同一 DAG 中使用 TaskGroup 和 PythonBranchOperator

Airflow - Use TaskGroup and PythonBranchOperator in the same DAG

airflow-scheduler