PySpark多Executor场景下调用HTTP API如何遵守请求频次配额？

针对多Executor并行调用受限API的场景，以下是几个可落地的解决方案：

1. 基于Redis的全局速率控制 #

利用Redis的原子操作实现跨Executor的全局请求计数和限流，确保所有节点的请求总和不超过每分钟1000次的配额。

实现思路 #

• 用Redis的INCR命令原子递增请求计数器，同时通过EXPIRE设置计数器60秒自动过期（实现每分钟重置）。
• 每个请求前先尝试递增计数器，若结果超过配额则等待后重试；在配额内则发起API请求。
• 结合原子操作避免多节点同时刷新计数器的冲突。

代码示例 #

import redis
import time
import requests
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 初始化Redis客户端（确保所有Executor可访问该实例）
redis_client = redis.Redis(host="redis-host", port=6379, db=0)

def api_call_with_rate_control(record_id):
    api_url = f"https://target-api.com/data/{record_id}"
    while True:
        # 原子递增计数器，首次创建时设置60秒过期
        current_count = redis_client.incr("api_request_count")
        if current_count == 1:
            redis_client.expire("api_request_count", 60)
        
        if current_count <= 1000:
            try:
                response = requests.get(api_url)
                response.raise_for_status()
                return str(response.json())
            except requests.exceptions.HTTPError as e:
                if response.status_code == 429:
                    time.sleep(2)
                    continue
                raise e
        else:
            time.sleep(1)
            continue

# 注册UDF
api_udf = udf(api_call_with_rate_control, StringType())
# 处理数据
df = df.withColumn("api_data", api_udf(df["record_id"]))

2. 分区级限速 + 指数退避重试 #

将数据分区后，给每个分区分配固定配额，同时用指数退避策略处理429错误，避免集中触发限流。

实现思路 #

• 根据总配额和分区数，计算每个分区的每分钟请求上限（例如1000配额/10分区=100次/分区/分钟）。
• 在分区内部维护请求计数器，每发起一个请求后根据速率要求添加延迟。
• 用指数退避处理429错误，降低重复触发限流的概率。

代码示例 #

import requests
import time
from tenacity import retry, stop_after_attempt, wait_exponential
from pyspark.sql.types import StringType

# 全局配置
TOTAL_QUOTA_PER_MINUTE = 1000

def process_partition(partition):
    # 通过广播变量传递实际分区总数，此处为示例值
    total_partitions = 10
    partition_quota = TOTAL_QUOTA_PER_MINUTE // total_partitions
    request_count = 0
    start_time = time.time()
    
    @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=10))
    def call_api(record_id):
        nonlocal request_count, start_time
        current_time = time.time()
        # 每分钟重置计数器
        if current_time - start_time >= 60:
            request_count = 0
            start_time = current_time
        
        # 控制分区内请求速率
        if request_count >= partition_quota:
            wait_time = 60 - (current_time - start_time)
            time.sleep(wait_time if wait_time > 0 else 1)
            request_count = 0
            start_time = time.time()
        
        request_count += 1
        response = requests.get(f"https://target-api.com/data/{record_id}")
        response.raise_for_status()
        return str(response.json())
    
    for record in partition:
        yield (record["record_id"], call_api(record["record_id"]))

# 用mapPartitions替代UDF，更灵活控制分区逻辑
result_rdd = df.rdd.mapPartitions(process_partition)
result_df = result_rdd.toDF(["record_id", "api_data"])

3. 集中式请求代理 #

部署一个独立的API代理服务，所有Executor的请求都转发到该代理，由代理统一处理速率控制。

实现思路 #

• 用Flask/FastAPI编写代理服务，通过ratelimit库控制每分钟1000次请求。
• Executor的UDF只需调用代理服务，无需关心限流逻辑，所有配额控制由代理统一处理。

代理服务代码（Flask示例） #

from flask import Flask, request
import requests
from ratelimit import limits, sleep_and_retry

app = Flask(__name__)
API_QUOTA = 1000
API_WINDOW = 60  # 秒

@sleep_and_retry
@limits(calls=API_QUOTA, period=API_WINDOW)
def forward_request(api_url):
    response = requests.get(api_url)
    return response.content, response.status_code

@app.route("/proxy")
def proxy():
    api_url = request.args.get("url")
    if not api_url:
        return "Missing 'url' parameter", 400
    content, status = forward_request(api_url)
    return content, status

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Executor端UDF代码 #

import requests
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def call_via_proxy(record_id):
    target_url = f"https://target-api.com/data/{record_id}"
    proxy_url = f"http://proxy-service:5000/proxy?url={target_url}"
    response = requests.get(proxy_url)
    response.raise_for_status()
    return str(response.json())

api_udf = udf(call_via_proxy, StringType())
df = df.withColumn("api_data", api_udf(df["record_id"]))

4. 预批次处理（适合小数据量） #

如果数据量不大，可将所有需要请求的记录收集到Driver，按时间窗口分批发起请求，再将结果合并回DataFrame。

代码示例 #

import requests
import time
from pyspark.sql import Row

# 收集所有待请求的记录ID
records = df.select("record_id").collect()
record_ids = [r.record_id for r in records]

# 按配额分批处理
batch_size = 1000
results = []

for i in range(0, len(record_ids), batch_size):
    batch = record_ids[i:i+batch_size]
    batch_results = []
    for record_id in batch:
        response = requests.get(f"https://target-api.com/data/{record_id}")
        response.raise_for_status()
        batch_results.append(Row(record_id=record_id, api_data=str(response.json())))
    results.extend(batch_results)
    # 非最后一批则等待60秒
    if i + batch_size < len(record_ids):
        time.sleep(60)

# 合并结果到原DataFrame
result_df = spark.createDataFrame(results)
final_df = df.join(result_df, on="record_id", how="left")

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