Airflow 2.3.4中如何为动态映射任务添加延迟

针对你遇到的400速率限制错误，这里提供几种实用的解决思路：

1. 基于任务索引实现递增延迟（精准控制间隔） #

通过TaskFlow API将BigQuery插入逻辑包装成自定义任务，利用Dynamic Task Mapping自带的map_index参数，为每个映射任务计算递增的延迟时间，让任务错开执行：

from airflow.decorators import task
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
import time

@task
def generate_configs():
    return [{...}]  # 你的5000个配置项

@task
def delayed_bq_insert(config, map_index):
    # 每个任务间隔1秒，可根据BigQuery配额调整
    delay_seconds = map_index * 1
    time.sleep(delay_seconds)
    
    # 执行BigQuery插入
    bq_operator = BigQueryInsertJobOperator(
        task_id=f"delayed_insert_{map_index}",
        configuration=config,
        gcp_conn_id="your_gcp_connection_id"  # 替换为你的GCP连接ID
    )
    bq_operator.execute(context=None)

configurations = generate_configs()
# 展开任务，传入配置和索引
delayed_bq_insert.expand(config=configurations, map_index=range(len(configurations)))

这种方式能确保任务按顺序错开请求，避免瞬间触发大量请求，但总执行时间会随任务数量和延迟时间增加。

2. 调整Airflow并行度限制（简化方案） #

直接在DAG定义中限制并发任务数量，降低单位时间内的请求频次，无需修改任务逻辑：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="your_bq_insert_dag",
    start_date=datetime(2024, 1, 1),
    concurrency=5,  # 同时运行的最大任务数
    max_active_tasks=5,  # DAG的最大活跃任务数
    schedule_interval=None
):
    # 你的任务定义
    configurations = generate_configs()
    delayed_bq_insert.expand(config=configurations, map_index=range(len(configurations)))

可根据BigQuery的实际配额调整concurrency值，比如设为10或20，平衡执行效率和请求速率。

3. 改用批量插入（最优效率方案） #

如果你的插入逻辑支持批量操作，将多个配置项合并为一个批量插入任务，直接减少请求次数，从根源上避免速率限制：

from airflow.decorators import task
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

@task
def generate_configs():
    return [{...}]  # 5000个配置项

@task
def batch_bq_insert(batch_configs):
    # 构造批量插入的配置（根据你的表结构调整字段定义）
    batch_insert_config = {
        "query": {
            "query": """
                INSERT INTO `your-project.your-dataset.your-table`
                VALUES UNNEST(@rows)
            """,
            "parameterMode": "NAMED",
            "queryParameters": [
                {
                    "name": "rows",
                    "parameterType": {
                        "type": "ARRAY",
                        "arrayElementType": {
                            "type": "STRUCT",
                            "structTypes": [
                                {"name": "column1", "type": "STRING"},
                                {"name": "column2", "type": "INTEGER"}
                                # 按需添加其他字段
                            ]
                        }
                    },
                    "parameterValue": {
                        "arrayValues": [{"structValues": cfg} for cfg in batch_configs]
                    }
                }
            ]
        }
    }
    
    bq_operator = BigQueryInsertJobOperator(
        task_id="batch_insert_task",
        configuration=batch_insert_config,
        gcp_conn_id="your_gcp_connection_id"
    )
    bq_operator.execute(context=None)

configurations = generate_configs()
# 将配置按每组50个拆分（可根据配额调整批量大小）
batches = [configurations[i:i+50] for i in range(0, len(configurations), 50)]
batch_bq_insert.expand(batch_configs=batches)

这种方式能将5000次请求压缩为100次（按每组50个计算），大幅提升执行效率，是优先推荐的方案。

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