Задачи уже зарегистрировали ошибку при динамической установке зависимостей

Question

Я пытаюсь динамически установить зависимости и получить task_0 уже зарегистрированным. Ниже приведен код.

with dag:

    final_task = DummyOperator(task_id='final')

    for i in range(0, 3):
        d1 = DummyOperator(task_id='tasks_{0}'.format(i))

        for j in range(0, 3):
            d2 = PythonOperator(task_id='task_{0}'.format(i),
                            python_callable=test_callable,
                            provide_context=True)
            d1 >> d2

Answer 1

1) Где вы определили d1?

2) Когда i = 0, создается 3 PythonOperator с тем же task_id, что составляет task_0 внутри цикла for для j.Вот почему вы видели это сообщение об ошибке.