R语言传入列表到函数调用TfL API批量生成汇总df的方法

核心优化逻辑 #

现有单条查询函数api_function已支持单个行程参数输入，批量场景需围绕限流控制、并发效率、异常容错、内存友好四个维度调整，规避串行耗时过长、触发API封禁、循环拼接DataFrame效率极低、单条错误导致全量任务中断的常见问题。

具体实现步骤 #

• 标准化单条查询函数的输入输出
  • 调整api_function仅接收单个行程参数（即当前传入taxi.args.new[0]的入参），移除内部绑定列表索引的逻辑，保证传入任意单条参数即可返回对应结果
  • 统一输出的DataFrame列结构，至少包含行程参数、行程段数、步行时长、总耗时、查询错误标记字段，无有效结果时返回带错误标记的同结构空DataFrame，不直接抛异常中断流程
• 适配TfL API的速率限制
  • 通过频率控制逻辑匹配TfL的API调用配额，比如免费版通常限制每秒10次请求，每发起10个请求就强制等待1秒，避免无限制高频请求触发IP封禁
  • 并发数控制在5-10区间即可，过高并发反而容易触发限流规则
• 分批拆分长列表，避免内存溢出
  • 不要一次性加载20万条全量参数到处理队列，按每批1000-2000条拆分为独立小批次逐批处理
  • 禁止在循环内反复concat DataFrame：循环内逐行拼接的时间复杂度为O(n²)，20万条数据会直接导致程序卡死；所有单条结果先存入列表，全部批次处理完成后一次性调用pd.concat()生成最终汇总df，效率可提升数个数量级
• 增加全局异常捕获逻辑
  • 单条查询调用外层套try-except逻辑，遇到请求超时、参数无效、API返回报错的场景，直接标记该条为查询失败并记录错误原因，不中断全量批量任务
• 可选优化：实现断点续跑
  • 每完成一个批次的处理，就将该批次的结果存为本地临时parquet文件；如果中途程序崩溃，下次启动可直接跳过已处理的批次，无需从头重跑

示例代码结构 #

import pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# 改造后的单条查询函数
def api_function(single_trip_param):
    try:
        # 原有TfL API调用、数据解析逻辑
        raw_trip_data = call_tfl_api(single_trip_param)
        # 解析为固定结构单条df的逻辑
        return parsed_single_result_df
    except Exception as e:
        # 出错返回同结构带错误标记的df，列与正常返回完全对齐
        return pd.DataFrame([{"trip_param": single_trip_param, "error": str(e)}])

# 批量处理主逻辑
def batch_process(trip_param_list, batch_size=1000, max_workers=5, req_per_sec=10):
    all_results = []
    # 按批次拆分全量参数列表
    for batch_start in range(0, len(trip_param_list), batch_size):
        current_batch = trip_param_list[batch_start : batch_start + batch_size]
        batch_result_list = []
        # IO密集型的API调用用线程池，比进程池开销低3-5倍
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # 提交当前批次所有查询任务
            future_map = [executor.submit(api_function, param) for param in current_batch]
            for req_idx, future in enumerate(as_completed(future_map)):
                batch_result_list.append(future.result())
                # 限流控制：每完成req_per_sec个请求等待1秒
                if (req_idx + 1) % req_per_sec == 0:
                    time.sleep(1)
        # 把当前批次结果加入总结果列表
        all_results.extend(batch_result_list)
        # 可选：每批存临时文件，支持断点续跑
        pd.concat(batch_result_list).to_parquet(f"temp_batch_{batch_start}.parquet", index=False)
    # 全量结果一次性拼接
    final_summary_df = pd.concat(all_results, ignore_index=True)
    return final_summary_df

# 调用方式
if __name__ == "__main__":
    full_trip_list = taxi.args.new
    final_df = batch_process(full_trip_list)

关键注意事项 #

• API调用属于IO密集型任务，不要使用多进程处理，线程池的启动开销更低，更适合高并发请求场景
• 提前评估内存占用：20万条解析后的结果如果每条平均占1KB，总内存占用约200MB，普通办公机即可承载
• 固定TfL API的鉴权请求头，不要在每次请求时重复生成鉴权信息，减少额外开销

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