影刀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条为一批)。批次间留间隔、断点要记录、错误单批隔离。
作者:林焱