使用Airflow的load_table_from_uri时动态通配符失效问题

我尝试在Airflow中使用load_table_from_uri API将多个GCS文件加载到BigQuery，但动态生成的URI无法正常工作，相同格式的硬编码URI却能正常运行。

动态生成URI的代码 #

uri="gs://b5a34db6ab213379-eu-pm-test-uat-data-ingest-temp/test_pm_cm_data/"+str(run_id).strip()+"/*.csv"

可正常运行的硬编码示例 #

uri="gs://b5a34db6ab213379-eu-pm-test-uat-data-ingest-temp/test_pm_cm_data/815756c6-19da-45cb-80cb-1d32645f223e/*.csv"

self.bq_load(uri,table_name)

def bq_load(self, uri,table_id):
    print(f"Loading {uri} to {table_id}")
    client = bigquery.Client(project="bmas-eu-pm-test-uat-data")
    job_config = bigquery.LoadJobConfig(
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    autodetect=True,
    allow_jagged_rows=True,
    allow_quoted_newlines=True,
    source_format=bigquery.SourceFormat.CSV,
    field_delimiter='\t' if uri.endswith('tsv') else ','
    )
    load_job = client.load_table_from_uri(
        uri, table_id, job_config=job_config
    )
    state = load_job.result().state
    print(f"load_job.state={state}")

报错信息 #

google.api_core.exceptions.NotFound: 404 Not found: Uris 

排查与解决方法 #

• 验证run_id的实际值
  打印str(run_id).strip()的输出，确认生成的URI和硬编码版本完全一致。Airflow的run_id可能带有manual__这类前缀，如果目标GCS路径没有对应前缀，就会导致路径不存在。

• 确认GCS路径存在
  用gsutil命令检查动态生成的路径是否有文件：

  gsutil ls gs://b5a34db6ab213379-eu-pm-test-uat-data-ingest-temp/test_pm_cm_data/[实际run_id值]/*.csv
  

  如果命令返回文件列表，说明路径没问题；反之则需要确认run_id对应的GCS目录是否正确生成。

• 优化字符串拼接方式
  改用f-string拼接URI，减少拼接错误：

  uri = f"gs://b5a34db6ab213379-eu-pm-test-uat-data-ingest-temp/test_pm_cm_data/{str(run_id).strip()}/*.csv"
  

• 检查权限与执行上下文
  确保Airflow Worker使用的服务账号拥有该GCS路径的读取权限；同时确认任务实例中获取的run_id与DAG运行时的实际run_id一致。

内容的提问来源于stack exchange，提问作者Vikrant Singh Rana