BranchPythonOperator之后的任务意外跳过

2024-03-28 17:00:23 发布

您现在位置:Python中文网/ 问答频道 /正文

我的dag定义如下。尽管flag1flag2都是y,但它们还是被跳过了。你知道吗

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import BranchPythonOperator
import pandas as pd
from itertools import compress



default_args = {
    'owner': 'alex'
    , 'retries': 2
    , 'retry_delay': timedelta(minutes=15)
    , 'depends_on_past': False
    , 'start_date': datetime(2018, 11, 22)
}

dag = DAG(
    'test_dag'
    , catchup = False
    , default_args = default_args
    , schedule_interval = '@daily'
)

task1 = DummyOperator(
        task_id='task1',
        dag=dag,
    )

task2 = DummyOperator(
        task_id='task2',
        dag=dag,
    )

task3 = DummyOperator(
        task_id='task3',
        dag=dag,
    )


# 1 means yes, 0 means no
flag1 = 'y'
flag2 = 'y'
flag3 = 'y'

tasks_name = ['task1', 'task2', 'task3']
flags = [flag1, flag2, flag3]


def generate_branches(tasks_name, flags):
    res = []
    idx = 1
    root_name = 'switch'
    for sub_task, sub_flag in zip(tasks_name, flags):
        tmp_branch_operator = BranchPythonOperator(
            task_id=root_name+str(idx), # switch1, switch2, ...
            python_callable= lambda: sub_task if sub_flag == 'y' else 'None',
            dag=dag,
        )
        res.append(tmp_branch_operator)
        idx += 1
    return res


def set_dependencies(switches, transfer_operators):
    for sub_switch, sub_transfer_operator in zip(switches, transfer_operators):
        sub_switch.set_downstream(sub_transfer_operator)


transfer_operators = [task1, task2, task3]
gen_branches_op = generate_branches(tasks_name, flags)
set_dependencies(gen_branches_op, transfer_operators)

enter image description here


Tags: namefromimportidtaskoperatortransfertasks
1条回答
网友
1楼 · 发布于 2024-03-28 17:00:23

这个问题是由lambda的后期绑定行为引起的。因为lambda是在调用时计算的,所以每次lambda总是返回列表中的最后一个元素,即task3。你知道吗

如果您可以查看switch1和switch2的日志,您会发现它们分别有以下分支task3,而不是task1task2。你知道吗

为了避免这种情况,可以通过更改generate_branches()中的python_callable来强制在定义lambda时对其求值:

def generate_branches(tasks_name, flags):
    res = []
    idx = 1
    root_name = 'switch'
    for sub_task, sub_flag in zip(tasks_name, flags):
        tmp_branch_operator = BranchPythonOperator(
            task_id=root_name+str(idx), # switch1, switch2, ...
            python_callable=lambda sub_task=sub_task: sub_task if sub_flag == "y", else "None"
            dag=dag,
        )
        res.append(tmp_branch_operator)
        idx += 1
    return res

相关问题 更多 >