0

所以我在任务组中创建任务，并试图将它们添加到我的任务序列中，但它抛出了这个错误：

Broken DAG: [/Users/abc/projects/abc/airflow_dags/dag.py] Traceback (most recent call last):
  File "/Users/abc/.pyenv/versions/3.8.12/envs/vmd-3.8.12/lib/python3.8/site-packages/airflow/models/baseoperator.py", line 1234, in set_downstream
    self._set_relatives(task_or_task_list, upstream=False)
  File "/Users/abc/.pyenv/versions/3.8.12/envs/vmd-3.8.12/lib/python3.8/site-packages/airflow/models/baseoperator.py", line 1178, in _set_relatives
    task_object.update_relative(self, not upstream)
AttributeError: 'NoneType' object has no attribute 'update_relative'

我正在创建我的任务组和这样的任务：

def get_task_group(dag, task_group):
    t1 = DummyOperator(task_id='t1', dag=dag, task_group=task_group)
    t2 = DummyOperator(task_id='t2', dag=dag, task_group=task_group)
    t3 = DummyOperator(task_id='t3', dag=dag, task_group=task_group)
    t4 = DummyOperator(task_id='t4', dag=dag, task_group=task_group)
    t5 = DummyOperator(task_id='t5', dag=dag, task_group=task_group)
    t_list = [t2, t3, t4]
    t1.set_downstream(t_list)
    t5.set_upstream(t_list)


with DAG('some_dag', default_args=args) as dag:
    with TaskGroup(group_id=f"run_model_tasks", dag=dag) as tg:
      run_model_task_group = get_task_group(dag, tg)

    a1 = DummyOperator(task_id='a1', dag=dag)
    a2 = DummyOperator(task_id='a2', dag=dag)
    a3 = DummyOperator(task_id='a3', dag=dag)
    a4 = DummyOperator(task_id='a4', dag=dag)

    a1.set_downstream(a2)
    a2.set_downstream(run_model_task_group)
    a3.set_upstream(run_model_task_group)
    a3.set_downstream(a4)

如果我删除任务组并通过删除行将任务组任务从排序中退出

a2.set_downstream(run_model_task_group)
a3.set_upstream(run_model_task_group)

我可以看到 a1、a2、a3 和 a4 的顺序正确，并且我可以断开断开的run_model_task_group任务，但是只要将其添加到顺序中，就会出现上述错误。

谁能指导我这里可能发生的事情？ 在此处输入图像描述

请注意，我使用函数获取dag和task_group参数来创建任务组任务，因为我也想为另一个 dag 创建相同的任务集。

Python Version: 3.8.8
Airflow Version: 2.0.1
4

1 回答 1

1

AttributeError: 'NoneType' object has no attribute 'update_relative'发生这种情况是因为run_model_task_group它None超出了With块的范围，这是 Python 的预期行为。

在不改变到目前为止所做的事情的情况下，您可以重构get_task_group()以返回一个TaskGroup对象，如下所示：

def get_task_group(dag, group_id):

    with TaskGroup(group_id=group_id, dag=dag) as tg:
        t1 = DummyOperator(task_id='t1', dag=dag)
        t2 = DummyOperator(task_id='t2', dag=dag)
        t3 = DummyOperator(task_id='t3', dag=dag)
        t4 = DummyOperator(task_id='t4', dag=dag)
        t5 = DummyOperator(task_id='t5', dag=dag)
        t_list = [t2, t3, t4]
        t1.set_downstream(t_list)
        t5.set_upstream(t_list)
    return tg

在 DAG 定义中，只需调用它：

run_model_task_group = get_task_group(dag, "run_model_tasks")

生成的图形视图如下所示：

图视图

DAG 定义：

with DAG('some_dag',
         default_args=default_args,
         start_date=days_ago(2),
         schedule_interval='@once') as dag:
    # with TaskGroup(group_id=f"run_model_tasks", dag=dag) as tg:
    #     run_model_task_group = get_task_group(dag, )
    run_model_task_group = get_task_group(dag, "run_model_tasks")
    a1 = DummyOperator(task_id='a1', dag=dag)
    a2 = DummyOperator(task_id='a2', dag=dag)
    a3 = DummyOperator(task_id='a3', dag=dag)
    a4 = DummyOperator(task_id='a4', dag=dag)

    a1.set_downstream(a2)
    a2.set_downstream(run_model_task_group)
    a3.set_upstream(run_model_task_group)
    a3.set_downstream(a4)

最后，考虑使用按位运算符而不是set_downstreamand ，这是推荐的方式，而且这里的源代码set_upstream也不那么冗长。

让我知道这是否对你有用。

测试：气流版本：2.1.4，Python 3.8.10

于 2021-10-29T11:52:36.390 回答