影刀RPA 数据分页与分批处理：一次处理一万条不卡死
采集了一万条数据，然后全塞进一个ForEach里逐条处理——结果流程越来越慢，最后卡死。不是数据量的问题，是处理方式的问题。
这篇文章讲数据分批处理的正确姿势。
什么时候需要分批
你的数据量没过500条时，一次性处理没问题。但以下情况必须分批：
- 每条数据处理涉及网页操作（打开详情页、点击按钮）。网页操作本身就慢，一万条就是一万次页面交互。
- 每条数据处理涉及API调用。API有频率限制，一口气发10000个请求，不是被限流就是被封IP。
- 处理结果要实时写入文件。频繁的小写入远比批量写入慢。
- 内存不够。全量加载一万条数据到变量没问题，但如果每条的详情还有嵌套的子数据，内存就危险了。
方式一：按固定批次大小处理
# 把一个大列表切成每500个一批defchunk_list(data,batch_size=500):"""将列表按固定大小切分"""foriinrange(0,len(data),batch_size):yielddata[i:i+batch_size]total_records=10000all_data=[f'record_{i}'foriinrange(total_records)]batch_size=500processed=0forbatchinchunk_list(all_data,batch_size):# 处理当前批次forrecordinbatch:# 处理单条数据...passprocessed+=len(batch)print(f'进度：{processed}/{total_records}({processed/total_records*100:.1f}%)')# 批次间休息time.sleep(1)# 给系统一点喘息时间拼多多店群自动化上架方案
方式二：流式读取大文件
如果源数据是一个大Excel或CSV，用pandas的chunksize参数一批一批读，不用全部加载到内存：
importpandasaspd# CSV分批读取total=0forchunkinpd.read_csv('huge_file.csv',chunksize=1000,dtype=str):# chunk 是一个1000行的DataFramefor_,rowinchunk.iterrows():# 处理每一行passtotal+=len(chunk)print(f'已处理{total}行')对于Excel文件，pandas不支持直接按chunksize读取。替代方案是分Sheet读取，或者先转成CSV再分批。
用openpyxl逐行读取大Excel
fromopenpyxlimportload_workbook wb=load_workbook('large.xlsx',read_only=True)ws=wb.active batch=[]batch_size=500count=0forrowinws.iter_rows(values_only=True):batch.append(row)iflen(batch)>=batch_size:# 处理这一批process_batch(batch)count+=len(batch)print(f'已处理{count}行')batch=[]# 处理最后一批ifbatch:process_batch(batch)wb.close()read_only=True是关键参数——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_file=r'C:\data\checkpoint.json'# 读取上次进度try:withopen(checkpoint_file,'r')asf:checkpoint=json.load(f)start_batch=checkpoint.get('last_batch',0)except:start_batch=0batches=list(chunk_list(all_data,500))foriinrange(start_batch,len(batches)):batch=batches[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':i+1,'total':len(batches)},f)print(f'批次{i+1}/{len(batches)}完成')exceptExceptionase:print(f'批次{i+1}失败：{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条为一批）。批次间留间隔、断点要记录、错误单批隔离。
作者：林焱
