Airflow中基于BigQueryGetDataOperator的动态任务映射失败求助

错误原因 #

1. BigQuery返回格式不匹配：你注释掉了as_dict=True，BigQueryGetDataOperator默认返回二维列表（每行是字段值的列表，比如[[project1], [project2], ...]），而非字典列表。
2. 任务映射参数传递逻辑错误：expand(op_args=XComArg(get_data))会把get_data返回的列表中每个元素单独传给printer_task，也就是说每个映射任务拿到的是单个[project_id]列表，而非完整数据列表。
3. 任务处理逻辑错误：你在printer_task里循环传入的参数，把每个元素当成字典调用get('project')，但实际循环[project_id]时，value是字符串类型，自然没有get方法，触发报错。

解决方案 #

步骤1：修正BigQuery返回格式 #

取消as_dict=True的注释，让返回结果为字典列表，每个字典包含project键：

def fetch_data_from_bq():
    return BigQueryGetDataOperator(
        task_id="fetch_data_from_bq",
        project_id="Project-X",
        dataset_id="Dataset-X",
        table_id="Table-X",
        gcp_conn_id="bigquery_default",
        max_results=10,
        selected_fields='project',
        as_dict=True  # 启用字典格式返回
    )

步骤2：调整任务处理逻辑与映射方式 #

因为每个映射任务只需要处理单个project数据，修改printer_task接收单个字典参数：

def printer_task(project_dict):
    project_id = project_dict.get('project')
    print(f"project_ID: {project_id}")

保持映射代码不变，此时expand会自动将字典列表中的每个元素传给对应任务：

create_mapped_tasks = PythonOperator.partial(
    task_id="create_mapped_tasks",
    python_callable=printer_task
).expand(op_args=XComArg(get_data))

步骤3：补充缺失的DAG导入 #

你的代码中使用了models.DAG但未导入，需补充：

from airflow.models import DAG

完整修正代码 #

import datetime
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryGetDataOperator
from airflow import XComArg

# 补充定义default_args（示例）
default_args = {
    'owner': 'airflow',
    'start_date': datetime.datetime(2024, 2, 1),
}

def printer_task(project_dict):
    project_id = project_dict.get('project')
    print(f"project_ID: {project_id}")

def fetch_data_from_bq():
    return BigQueryGetDataOperator(
        task_id="fetch_data_from_bq",
        project_id="Project-X",
        dataset_id="Dataset-X",
        table_id="Table-X",
        gcp_conn_id="bigquery_default",
        max_results=10,
        selected_fields='project',
        as_dict=True
    )

with DAG(
        dag_id="dummy-task",
        default_args=default_args,
        schedule_interval='@daily',
) as dag:

    get_data = fetch_data_from_bq()
    start_task = DummyOperator(task_id="start_task")
    end_task = DummyOperator(task_id="end_task")

    create_mapped_tasks = PythonOperator.partial(
        task_id="create_mapped_tasks",
        python_callable=printer_task
    ).expand(op_args=XComArg(get_data))

    start_task >> get_data >> create_mapped_tasks >> end_task

内容的提问来源于stack exchange，提问作者Hürkan Utan