对象的气流分支错误不可迭代

Question

我正在尝试让 BranchPythonOperator 工作，但出现以下错误：

'BigQueryInsertJobOperator' object is not iterable

这是我的分支操作员：

branching = BranchPythonOperator(
        task_id='branching',
        python_callable=return_branch,
        provide_context=True)

这是我的 Python 可调用对象：

def return_branch(ds, **kwargs):
    execution_year = kwargs['execution_date'].strftime("%Y")
    type = dataset_metadata[f'{execution_year}']['var']
    if type == 'foo':
        return x
    return y

x 和 y 是 BigQueryInsertJobOperator：

x = BigQueryInsertJobOperator(
    task_id='x',
    configuration={
        "query": {
            "query": "{% include 'q.sql' %}",
            "use_legacy_sql": False
            }
        },
    dag=dag)

Answer 1

我想参考this answer。您的方法return_branch 不应返回运算符。它必须返回您的运营商的task_id。你会得到这样的东西：

def return_branch(ds, **kwargs):
    next_task_id = "a" # <some kind of logic>

    return next_task_id


branching = BranchPythonOperator(
    task_id="pick_query",
    python_callable=return_branch,
    provide_context=True,
)

option_1 = DummyOperator(task_id="a")
option_2 = DummyOperator(task_id="b")

branching >> [option_1, option_2]