Airflow中BigQueryToBigQueryOperator复制BQ表失败求助

以下是你的代码需要修改的几个关键地方：

• 表名参数不完整：source_project_dataset_tables和destination_project_dataset_table必须填写完整的表名，格式为项目ID.数据集ID.表名，你现在只写到了数据集层级，缺少具体表名，这会导致Operator找不到要复制的对象。
• bigquery_conn_id参数配置错误：这个参数是Airflow中预配置的BigQuery连接ID（比如google_cloud_default），而不是项目ID.数据集格式的字符串。由于你已经指定了gcp_conn_id='google_cloud_default'，可以直接删除bigquery_conn_id这个参数，避免冲突。
• 建议补充location参数：如果你的BigQuery数据集不在默认的US多区域，必须显式指定location参数（比如location='asia-east1'），否则会因区域不匹配报错。
• 权限验证：确保impersonation_chain指定的服务账号拥有源表的bigquery.tables.getData权限，以及目标数据集的bigquery.tables.create和bigquery.tables.updateData权限。

修正后的完整代码： #

from airflow import DAG
from astro_plugins.sensors.bigquery_sql_sensor import BigquerySqlSensor
from airflow.providers.google.cloud.transfers.bigquery_to_bigquery import BigQueryToBigQueryOperator
from dags.config import var

import datetime as dt

dag = DAG(
    dag_id="dag_copy_bq_to_bq",
    default_args={
        "owner": "<owner>",
        "email": ["our_email@email.com"],
        "email_on_failure": True,
        "email_on_retry": False,
        "depends_on_past": False,
        "start_date": dt.datetime(2023, 9, 19, 7, 0, 0),
        "retries": 0,
    },
    schedule_interval="@hourly",
)

copy_table_task = BigQueryToBigQueryOperator(
    task_id='copy_table_task',
    # 补充完整表名：项目ID.源数据集.源表名
    source_project_dataset_tables='<project_id>.<source_dataset>.<source_table>',
    # 补充完整表名：项目ID.目标数据集.目标表名
    destination_project_dataset_table='<project_id>.<destination_dataset>.<destination_table>',
    write_disposition='WRITE_APPEND',
    create_disposition='CREATE_IF_NEEDED',
    schema_update_options=['ALLOW_FIELD_ADDITION', 'ALLOW_FIELD_RELAXATION'],
    gcp_conn_id='google_cloud_default',
    # 补充数据集所在区域，根据实际情况修改
    location='asia-east1',
    impersonation_chain=var("astro_team_sa")
)

内容的提问来源于stack exchange，提问作者Radka Žmers