分块调用API收集CU ID后，如何一次性执行机器学习预测任务？

问题背景 #

因CU ID数量过多会导致URL过长报错，所以采用分块调用API的方式，但当前每次调用都会立即执行机器学习预测并推送结果至数据库，运行速度较慢。需要保留分块调用的方式，让API仅存储每次传来的CU ID，待所有调用完成后再一次性执行预测及入库操作。

解决方案 #

当然可以实现，核心思路是让API分阶段处理：先收集所有CU ID，再触发批量执行。以下是具体修改方案：

1. 服务端（API）修改 #

新增存储CU ID的容器，以及触发批量执行的专用接口：

# 全局存储收集到的所有CU ID（若为多进程/多实例部署，建议改用Redis、数据库等分布式存储）
collected_cu_ids = []

@app.route('/collect-cu', methods=['GET'])
def collect_cu():
    cu_string = request.args.get('CU', type=str)
    if not cu_string:
        return jsonify({"error": "CU参数不能为空"}), 400
    cu_list = cu_string.split(',')
    collected_cu_ids.extend(cu_list)
    return jsonify({"collected_count": len(collected_cu_ids)})

@app.route('/process-all', methods=['POST'])
def process_all():
    if not collected_cu_ids:
        return jsonify({"error": "未收集到任何CU ID"}), 400
    # 执行批量预测与入库
    jm.create_job_CU(collected_cu_ids)
    results = [job.result for job in jm.completed_jobs]
    # 清空存储，避免重复执行
    collected_cu_ids.clear()
    jm.completed_jobs = []
    return jsonify(results)

2. 客户端修改 #

先分块调用收集接口完成所有CU ID上传，再触发批量处理：

# 分块上传所有CU ID
for chunk in cu_chunks:
    cu_string = ','.join(chunk)
    url = f'{base_url}/collect-cu?CU={cu_string}'
    response = requests.get(url)
    print(f'上传{len(chunk)}个CU ID，状态码: {response.status_code}，已累计收集: {response.json()["collected_count"]}')

# 所有块上传完成后，触发批量处理
process_url = f'{base_url}/process-all'
response = requests.post(process_url)
print(f'批量处理完成，状态码: {response.status_code}')
print('处理结果:', response.json())

注意事项 #

• 若API为多进程/多实例部署，全局列表collected_cu_ids会失效，必须改用Redis、数据库等分布式存储保存CU ID。
• 可给收集接口添加任务ID参数，支持同时处理多个客户端的收集请求，避免不同任务的CU ID混淆。
• 建议增加超时清理机制，防止未触发执行的CU ID长期占用存储资源。

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