python之在 Airflow 中生成多个任务时颠倒上游/下游关系

这道题相关的原代码可以找here .

我对 up 位移运算符和 set_upstream/set_downstream 方法在我的 DAG 中定义的任务循环中工作感到困惑。当DAG的主执行循环配置如下:

for uid in dash_workers.get_id_creds(): 
    clear_tables.set_downstream(id_worker(uid)) 

或

for uid in dash_workers.get_id_creds(): 
    clear_tables >> id_worker(uid) 

图表看起来像这样(字母数字序列是用户 ID，它还定义了任务 ID):

当我像这样配置 DAG 的主执行循环时:

for uid in dash_workers.get_id_creds(): 
    clear_tables.set_upstream(id_worker(uid)) 

或

for uid in dash_workers.get_id_creds(): 
    id_worker(uid) >> clear_tables 

图表看起来像这样:

第二张图是我想要的/我期望根据我阅读文档生成的前两个代码片段。如果我希望 clear_tables 在为不同用户 ID 触发我的批数据解析任务之前首先执行，我应该将其指示为 clear_tables >> id_worker(uid)

编辑 -- 这是完整的代码，自从我发布最后几个问题以来已经更新，以供引用:

from datetime import datetime 
import os 
import sys 
 
from airflow.models import DAG 
from airflow.operators.python_operator import PythonOperator 
 
import ds_dependencies 
 
SCRIPT_PATH = os.getenv('DASH_PREPROC_PATH') 
if SCRIPT_PATH: 
    sys.path.insert(0, SCRIPT_PATH) 
    import dash_workers 
else: 
    print('Define DASH_PREPROC_PATH value in environmental variables') 
    sys.exit(1) 
 
ENV = os.environ 
 
default_args = { 
  'start_date': datetime.now(), 
} 
 
DAG = DAG( 
  dag_id='dash_preproc', 
  default_args=default_args 
) 
 
clear_tables = PythonOperator( 
  task_id='clear_tables', 
  python_callable=dash_workers.clear_db, 
  dag=DAG) 
 
def id_worker(uid): 
    return PythonOperator( 
        task_id=id, 
        python_callable=dash_workers.main_preprocess, 
        op_args=[uid], 
        dag=DAG) 
 
for uid in dash_workers.get_id_creds(): 
    preproc_task = id_worker(uid) 
    clear_tables << preproc_task 

在实现@LadislavIndra 的建议后，我继续对位移运算符进行相同的反向实现，以获得正确的依赖关系图。

更新 @AshBerlin-Taylor 的回答是这里发生的事情。我假设 Graph View 和 Tree View 在做同样的事情，但事实并非如此。以下是 id_worker(uid) >> clear_tables 在图 TableView 中的样子:

我当然不希望我的数据预准备例程的最后一步是删除所有数据表!

请您参考如下方法:

Airflow 中的 TreeView 与您(和我!)最初的想法“倒退”。在您的第一个屏幕截图中，它显示“clear_tables”必须在“AAAG5608078M2”运行任务之前运行。 DAG 状态取决于每个 id worker 任务。因此，它不是任务顺序，而是状态链树。如果这有任何意义的话。

(乍一看这可能看起来很奇怪，但这是因为 DAG 可以分支出来并返回分支。)

查看 dag 的 Graph View 可能会更幸运。这个有箭头并以更直观的方式显示执行顺序。 (虽然我现在确实发现 TreeView 很有用。只是开始时不太清楚)