Airflow 如何在循环生成的串行依赖任务前添加前置任务

问题原因 #

现有代码存在两个问题导致依赖不符合预期：

• itemized_costs分支下循环生成8个串行任务时，循环内的s3变量会被逐次覆盖，循环结束后s3指向最后一个任务itemized_costs_to_S3_7，最终执行latest_only >> s3时，仅将最后一个任务挂到了前置节点下游，前7个任务未和latest_only建立依赖关系。
• 字符串值判断使用了is而非==，is校验的是对象内存地址而非值，可能出现分支判断失效的问题。

修复方案 #

针对itemized_costs分支，完成8个任务的串行串联后，直接将串行任务组的第一个任务设置为latest_only的下游即可，其余非分片任务的逻辑保持不变。修复后代码如下：

for endpoint in ENDPOINTS:
    latest_only = LatestOnlyOperator(
        task_id=f'{endpoint.name}_latest_only',
    )

    if endpoint.name == 'itemized_costs':
        task_list = []
        # 生成8个分片上传S3的任务并配置串行依赖
        for i in range(0, 8):
            s3_task = PToS3Operator(
                task_id=f'{endpoint.name}_to_S3_{i}',
                task_no=f'{int(i)}',
                pool_slots=5,
                endpoint=endpoint
            )
            task_list.append(s3_task)
            if i != 0:
                task_list[i-1] >> task_list[i]
        # 前置节点指向串行任务组的首个任务
        latest_only >> task_list[0]

    else:
        s3_task = PToS3Operator(
            task_id=f'{endpoint.name}_to_S3',
            pool_slots=5,
            endpoint=endpoint
        )
        latest_only >> s3_task

优化说明 #

• 调整了循环内变量命名，避免和外层分支变量同名导致的覆盖问题，降低后续维护的出错概率。
• 修正了字符串判断的语法问题，避免分支逻辑异常。
• 移除了未使用的空列表l，减少冗余代码。

内容的提问来源于stack exchange，提问作者KristiLuna