影刀RPA 数据分页与分批处理一次处理一万条不卡死采集了一万条数据然后全塞进一个ForEach里逐条处理——结果流程越来越慢最后卡死。不是数据量的问题是处理方式的问题。这篇文章讲数据分批处理的正确姿势。什么时候需要分批你的数据量没过500条时一次性处理没问题。但以下情况必须分批每条数据处理涉及网页操作打开详情页、点击按钮。网页操作本身就慢一万条就是一万次页面交互。每条数据处理涉及API调用。API有频率限制一口气发10000个请求不是被限流就是被封IP。处理结果要实时写入文件。频繁的小写入远比批量写入慢。内存不够。全量加载一万条数据到变量没问题但如果每条的详情还有嵌套的子数据内存就危险了。方式一按固定批次大小处理# 把一个大列表切成每500个一批defchunk_list(data,batch_size500):将列表按固定大小切分foriinrange(0,len(data),batch_size):yielddata[i:ibatch_size]total_records10000all_data[frecord_{i}foriinrange(total_records)]batch_size500processed0forbatchinchunk_list(all_data,batch_size):# 处理当前批次forrecordinbatch:# 处理单条数据...passprocessedlen(batch)print(f进度{processed}/{total_records}({processed/total_records*100:.1f}%))# 批次间休息time.sleep(1)# 给系统一点喘息时间拼多多店群自动化上架方案方式二流式读取大文件如果源数据是一个大Excel或CSV用pandas的chunksize参数一批一批读不用全部加载到内存importpandasaspd# CSV分批读取total0forchunkinpd.read_csv(huge_file.csv,chunksize1000,dtypestr):# chunk 是一个1000行的DataFramefor_,rowinchunk.iterrows():# 处理每一行passtotallen(chunk)print(f已处理{total}行)对于Excel文件pandas不支持直接按chunksize读取。替代方案是分Sheet读取或者先转成CSV再分批。用openpyxl逐行读取大Excelfromopenpyxlimportload_workbook wbload_workbook(large.xlsx,read_onlyTrue)wswb.active batch[]batch_size500count0forrowinws.iter_rows(values_onlyTrue):batch.append(row)iflen(batch)batch_size:# 处理这一批process_batch(batch)countlen(batch)print(f已处理{count}行)batch[]# 处理最后一批ifbatch:process_batch(batch)wb.close()read_onlyTrue是关键参数——openpyxl不会把整个文件加载到内存而是按需读取。处理10万行的Excel内存占用从几个G降到几十MB。方式三分段采集不是采集完再处理而是边采集边处理——采集一页立刻处理这一页的数据。流程 page 1 WHILE True: 【打开网页】→ 列表页?page{page} 【获取元素列表】→ 本页数据 IF 本页数据为空: BREAK # 后面没有了 # 立刻处理本页数据不等全部采集完 Python节点 → 处理 current_page_data # 或者保存本页到文件 【写入Excel】→ 追加模式写入或写新文件 page 1 【等待 2 秒】 # 分页间隔边采边处理的好处不怕流程中途崩溃已经处理过的数据不会丢内存占用小只保留当前页的数据可以通过断点续跑的思路续接方式四生产者-消费者模式采集是生产处理是消费。两者分开采集线程往队列里放处理线程从队列里取。fromqueueimportQueuefromthreadingimportThreadimporttime# 这个模式在影刀中实现比较复杂影刀的线程模型有限制# 简化为采集一批丢到文件 → 另一个流程读文件处理# 流程A采集# 采集完一批 → 写入 batch_001.xlsx# 继续采集下一批 → 写入 batch_002.xlsx# 流程B处理# 轮询文件夹 → 发现有新的 batch_xxx.xlsx# 读取处理 → 标记为已处理分批处理中的错误处理分批处理最怕的是第47批报错了前面46批白跑了。importjson checkpoint_filerC:\data\checkpoint.json# 读取上次进度try:withopen(checkpoint_file,r)asf:checkpointjson.load(f)start_batchcheckpoint.get(last_batch,0)except:start_batch0batcheslist(chunk_list(all_data,500))foriinrange(start_batch,len(batches)):batchbatches[i][video(video-7pxmwbWo-1784132807895)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]try:# 处理当前批次process_batch(batch)# 保存进度withopen(checkpoint_file,w)asf:json.dump({last_batch:i1,total:len(batches)},f)print(f批次{i1}/{len(batches)}完成)exceptExceptionase:print(f批次{i1}失败{e})print(f已处理到第{i}批可从断点续跑)break性能优化的几个技巧1. 减少循环内的文件读写。每次循环都打开→写入→关闭一个Excel跟循环外一次性写入性能差距巨大。# 慢循环内频繁写入forrecordindata:ws.append(record)wb.save(output.xlsx)# 不要在循环内save# 快循环外统一写入forrecordindata:ws.append(record)wb.save(output.xlsx)# 循环结束后统一save2. 用列表推导替代循环构建数据。# 慢result[]fordindata:result.append(clean(d))# 快result[clean(d)fordindata]3. 数据库批量插入。逐条INSERT和批量INSERT的性能差距可达100倍# 慢逐条插入forrecordindata:cursor.execute(INSERT INTO orders VALUES (%s, %s, %s),record)conn.commit()# 快批量插入cursor.executemany(INSERT INTO orders VALUES (%s, %s, %s),data)conn.commit()总结数据上千就分批、文件太大读流式、采完一页处理一页。核心原则——不要让任何单次操作处理超过你设的阈值建议500-1000条为一批。批次间留间隔、断点要记录、错误单批隔离。作者林焱