Python多进程分块读取超大文件的方法
作者:asdfsx 时间:2023-12-07 02:19:28
本文实例讲述了Python多进程分块读取超大文件的方法。分享给大家供大家参考,具体如下:
读取超大的文本文件,使用多进程分块读取,将每一块单独输出成文件
# -*- coding: GBK -*-
import urlparse
import datetime
import os
from multiprocessing import Process,Queue,Array,RLock
"""
多进程分块读取文件
"""
WORKERS = 4
BLOCKSIZE = 100000000
FILE_SIZE = 0
def getFilesize(file):
"""
获取要读取文件的大小
"""
global FILE_SIZE
fstream = open(file,'r')
fstream.seek(0,os.SEEK_END)
FILE_SIZE = fstream.tell()
fstream.close()
def process_found(pid,array,file,rlock):
global FILE_SIZE
global JOB
global PREFIX
"""
进程处理
Args:
pid:进程编号
array:进程间共享队列,用于标记各进程所读的文件块结束位置
file:所读文件名称
各个进程先从array中获取当前最大的值为起始位置startpossition
结束的位置endpossition (startpossition+BLOCKSIZE) if (startpossition+BLOCKSIZE)<FILE_SIZE else FILE_SIZE
if startpossition==FILE_SIZE则进程结束
if startpossition==0则从0开始读取
if startpossition!=0为防止行被block截断的情况,先读一行不处理,从下一行开始正式处理
if 当前位置 <=endpossition 就readline
否则越过边界,就从新查找array中的最大值
"""
fstream = open(file,'r')
while True:
rlock.acquire()
print 'pid%s'%pid,','.join([str(v) for v in array])
startpossition = max(array)
endpossition = array[pid] = (startpossition+BLOCKSIZE) if (startpossition+BLOCKSIZE)<FILE_SIZE else FILE_SIZE
rlock.release()
if startpossition == FILE_SIZE:#end of the file
print 'pid%s end'%(pid)
break
elif startpossition !=0:
fstream.seek(startpossition)
fstream.readline()
pos = ss = fstream.tell()
ostream = open('/data/download/tmp_pid'+str(pid)+'_jobs'+str(endpossition),'w')
while pos<endpossition:
#处理line
line = fstream.readline()
ostream.write(line)
pos = fstream.tell()
print 'pid:%s,startposition:%s,endposition:%s,pos:%s'%(pid,ss,pos,pos)
ostream.flush()
ostream.close()
ee = fstream.tell()
fstream.close()
def main():
global FILE_SIZE
print datetime.datetime.now().strftime("%Y/%d/%m %H:%M:%S")
file = "/data/pds/download/scmcc_log/tmp_format_2011004.log"
getFilesize(file)
print FILE_SIZE
rlock = RLock()
array = Array('l',WORKERS,lock=rlock)
threads=[]
for i in range(WORKERS):
p=Process(target=process_found, args=[i,array,file,rlock])
threads.append(p)
for i in range(WORKERS):
threads[i].start()
for i in range(WORKERS):
threads[i].join()
print datetime.datetime.now().strftime("%Y/%d/%m %H:%M:%S")
if __name__ == '__main__':
main()
希望本文所述对大家Python程序设计有所帮助。
标签:Python,多进程,读取文件
0
投稿
猜你喜欢
Python实现将蓝底照片转化为白底照片功能完整实例
2021-06-08 21:40:55
基于keras中的回调函数用法说明
2023-06-01 14:27:48
MySQL应用技巧之存取权限
2010-08-08 09:10:00
python-jwt用户认证食用教学的实现方法
2023-11-21 17:57:09
vite.config.js配置入门详解
2024-04-30 10:38:25
python内存动态分配过程详解
2023-10-02 14:58:46
Gradio机器学习模型快速部署工具应用分享前篇
2023-07-23 12:10:45
MySql如何实现远程登录MySql数据库过程解析
2024-01-28 06:47:51
PHP:微信小程序 微信支付服务端集成实例详解及源码下载
2023-11-14 13:37:55
设计能力决定权力
2009-06-16 14:48:00
【JS+CSS3】实现带预览图幻灯片效果的示例代码
2024-04-10 13:55:15
python Popen 获取输出,等待运行完成示例
2022-06-22 20:43:58
使用pandas读取csv文件的指定列方法
2023-07-07 13:11:26
Python网站验证码识别
2023-09-11 04:29:36
Python改变对象的字符串显示的方法
2022-11-10 17:16:12
python 环境安装及编辑器配置方法小结
2021-09-13 02:39:25
Python随机生成数据后插入到PostgreSQL
2023-05-14 19:58:09
python中如何实现径向基核函数
2023-11-28 02:48:45
python 多线程死锁问题的解决方案
2023-12-28 03:35:31
Python绘制词云图之可视化神器pyecharts
2023-02-23 01:35:25