使用GCSToBigQueryOperator从GCS加载数据到BQ时参数报错求助

错误原因 #

你遇到的参数错误，是因为GCSToBigQueryOperator不接受DAG作为参数，同时在with DAG(...) as dag:的上下文环境中，任务会自动关联到当前DAG，不需要手动传入DAG=dag这个参数，这属于无效传参。

修复后的代码 #

from airflow import DAG
from datetime import datetime,timedelta
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

default_args ={
        "retries": 1,
        "retry_delay": timedelta(minutes=5)
}
# 注意：不建议用datetime.today()作为start_date，每次DAG解析时日期会变化，建议用固定日期
with DAG(
    start_date= datetime(2024, 1, 1),
    dag_id="Gcs_to_bq",
    default_args=default_args,
    schedule_interval=None  # 无定时需求时显式指定，避免默认配置引发问题
) as dag:
 
load_from_gcs_to_bq = GCSToBigQueryOperator(
    task_id= "gcs_to_bq_land",
    bucket= "learn-composer",
    source_objects = ["landing_Employee_Ratings.csv"],
    destination_project_dataset_table= "uki-composer-sbox2-3350.composer.gcs-bq",
    skip_leading_rows=1,
    write_disposition="WRITE_TRUNCATE"
    # 移除无效的DAG=dag参数
)

额外提示 #

• start_date使用datetime.today()会导致每次DAG文件被解析时起始日期变为当天，可能触发Airflow调度逻辑异常，建议设置为固定的过去日期。
• 若DAG无需定时运行，显式设置schedule_interval=None更清晰，避免依赖Airflow默认调度配置。

内容的提问来源于stack exchange，提问作者DevX