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

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]