Блог

Я хочу вернуть 2 или более задач из функции, которые должны выполняться последовательно в том месте, где они вставлены в зависимости, см. Ниже.

 t1 = PythonOperator()

def generate_tasks():
    t2 = PythonOperator()
    t3 = PythonOperator()
    return magic(t2, t3) # magic needed here (preferably)

t1 >> generate_tasks() # otherwise here
# desired result: t1 >> t2 >> t3
 

Это выполнимо? Насколько я понимаю, Airflow 2.0, похоже, достигает этого с помощью TaskGroup, но мы используем Google Composer, и 2.0 некоторое время не будет доступен.

Лучшее решение, которое я нашел:

 t1 = PythonOperator()

def generate_tasks():
    t2 = PythonOperator()
    t3 = PythonOperator()
    return [t2, t3]

tasks = generate_tasks()
t1 >> tasks[0] >> tasks[1]
 

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

Как это сделать с помощью TaskGroup в Airflow 2.0:

 class Encryptor:
    def encrypt_and_archive(self):
        with TaskGroup("archive_and_encrypt") as section_1:
            encrypt = DummyOperator(task_id="encrypt")
            archive = BashOperator(task_id="archive", bash_command='echo 1')
            encrypt >> archive
        return section_1

with DAG(dag_id="example_return_task_group", start_date=days_ago(2), tags=["example"]) as dag:
    start = DummyOperator(task_id="start")
    encrypt_and_archive = Encryptor().encrypt_and_archive()
    end = DummyOperator(task_id='end')

             # 👇 single variable, containing two tasks
    start >> encrypt_and_archive >> end
 

Который создает следующий график:

Возможно ли что-то подобное удаленно до версии 2.0?

Вы не объяснили, что magic(t2, t3) это такое. TaskGroup — это строго функция пользовательского интерфейса, она не влияет на логику DAG. Согласно вашему описанию, кажется, что вы ищете определенную логику (иначе что это magic ?).

Я считаю, что это то, что вам нужно:

 default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 1, 24),
}
def generate_tasks():
    operator_list =[]
    for i in range(5): # Replace to generate the logic you wish to dynamically create tasks
        op = DummyOperator(task_id=f"t{str(i)}_task", dag=dag)
        if i>0:
            operator_list[i - 1] >> op
        operator_list.append(op)
    return operator_list

with DAG(
    dag_id='loop',
    default_args=default_args,
    schedule_interval=None,
) as dag:
    start_op = DummyOperator(task_id='start_task')
    end_op = DummyOperator(task_id='end_task')
    tasks = generate_tasks()
    start_op >> tasks[0]
    tasks[-1] >> end_op
 

Вы можете заменить DummyOperator любым оператором, который вы хотите.