Airflow并行写入BigQuery表触发403速率限制问题求助

问题场景 #

通过Airflow加载BigQuery表时遇到以下错误：

google.api_core.exceptions.Forbidden: 403 Exceeded rate limits: too many table update operations for this table

当前DAG并行处理20多张表，每张表对应的{table}_count任务都会将统计数据写入同一张BigQuery表my_table_in_bq，最终由Verify任务读取该表数据。核心任务定义代码如下：

def create_load_count_task(db_name, gcs_name, table_name):
    task = python_operator.PythonOperator(
        task_id=f'my_{table_name}_count',
        op_kwargs={
            'dataset_name': DATASET,
            'file_name' : f'{table_name}_metadata.json',
            'file_prefix': f'{gcs_name}',
            'table_name': f'my_table_in_bq',
            'table_load_type': bigquery.WriteDisposition.WRITE_APPEND,
        },           
        python_callable=load_into_bq
    )
    
    return task


with open(f'mypath/tables.conf') as fp:
    for count, line in enumerate(fp):
        config = line.split(':')

        db_name = config[0].strip()
        gcs_name = config[1].strip()
        table_name = config[2].strip()
 
        load = < my code >
        check = < my code >
        verify = < my code >
        init = < my code >
        load_count = create_load_count_task(db_name,gcs_name,table_name)

        print_dag_info >> check >> init >> load_count >> load >> verify

核心原因 #

BigQuery对单表的更新操作（包括WRITE_APPEND类型的写入）存在速率限制，20+并行任务同时往同一张表写入，超出了单表的操作频率阈值，触发403限流。

可行解决方案 #

1. 优化重试策略（比基础延迟重试更高效） #

仅用固定延迟重试可能效果有限，建议采用指数退避重试，配合Airflow的重试参数精准配置：

def create_load_count_task(db_name, gcs_name, table_name):
    task = python_operator.PythonOperator(
        task_id=f'my_{table_name}_count',
        op_kwargs={
            # 原有参数保留
            'dataset_name': DATASET,
            'file_name' : f'{table_name}_metadata.json',
            'file_prefix': f'{gcs_name}',
            'table_name': f'my_table_in_bq',
            'table_load_type': bigquery.WriteDisposition.WRITE_APPEND,
        },           
        python_callable=load_into_bq,
        # 重试配置
        retries=3,
        retry_delay=timedelta(seconds=10),
        retry_exponential_backoff=True,  # 开启指数退避，重试间隔依次为10s、20s、40s
        retry_on_exception=lambda e: isinstance(e, google.api_core.exceptions.Forbidden)
    )
    
    return task

同时在load_into_bq函数内部捕获403异常，确保Airflow能正确识别并触发重试逻辑。

2. 控制并行写入的并发数 #

避免20个任务同时写入同一张表，可通过以下方式精准控制：

• 使用任务池（Task Pool）：创建专属任务池，限制同时运行的写入任务数量：

# 在create_load_count_task中添加pool参数
task = python_operator.PythonOperator(
    # 其他参数不变
    pool='bq_write_pool',
    pool_slots=5  # 最多同时5个任务执行写入
)

之后在Airflow UI的「Admin → Pools」中创建bq_write_pool，设置Slots为5（可根据实际情况调整数值）。

• 调整DAG全局并行度：如果整个DAG的并行度过高，可在DAG定义中设置max_active_runs和concurrency参数，但此方式会影响DAG内所有任务，需谨慎使用。

3. 批量写入替代多次单表更新 #

如果每个load_count任务写入的数据量较小，可先将所有统计数据临时存储到GCS的统一路径下，再用单个任务一次性写入BigQuery：

1. 修改load_count任务，将统计数据写入GCS的临时文件（比如按表名拆分的独立文件，或追加到同一个聚合文件）；
2. 等待所有load_count任务完成后，新增batch_load_to_bq任务，读取GCS上的所有临时统计文件，一次性写入my_table_in_bq。

这种方式将多次单表更新合并为一次操作，从根源上规避限流问题。

4. 改用临时表合并方案 #

如果业务允许，可先将统计数据写入独立临时表，最后合并到目标表：

1. 每个load_count任务写入一张独立临时表（例如my_table_in_bq_temp_{table_name}）；
2. 所有临时表写入完成后，执行INSERT INTO my_table_in_bq SELECT * FROM my_table_in_bq_temp_*语句，完成数据合并。

此方案将写入压力分散到多个临时表，避免单表的速率限制触发。

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