我目前正在尝试读取一个大文件(8000万行),我需要为每个条目进行计算密集型矩阵乘法.计算完之后,我想将结果插入数据库.由于此过程采用时间密集的方式,我希望将文件拆分为多个核心以加快进程.
在研究之后,我发现了这个有希望的尝试,它将文件分成n个部分.
def file_block(fp,number_of_blocks,block):
'''
A generator that splits a file into blocks and iterates
over the lines of one of the blocks.
'''
assert 0 <= block and block < number_of_blocks
assert 0 < number_of_blocks
fp.seek(0,2)
file_size = fp.tell()
ini = file_size * block / number_of_blocks
end = file_size * (1 + block) / number_of_blocks
if ini <= 0:
fp.seek(0)
else:
fp.seek(ini-1)
fp.readline()
while fp.tell() < end:
yield fp.readline()
if __name__ == '__main__':
fp = open(filename)
number_of_chunks = 4
for chunk_number in range(number_of_chunks):
print chunk_number,100 * '='
for line in file_block(fp,number_of_chunks,chunk_number):
process(line)
虽然这有效,但我遇到了问题,使用多处理并行化:
fp = open(filename)
number_of_chunks = 4
li = [file_block(fp,chunk_number) for chunk_number in range(number_of_chunks)]
p = Pool(cpu_count() - 1)
p.map(processChunk,li)
虽然我理解这个错误,但是首先迭代整个文件以将所有行放入列表中是太昂贵了.
此外,我希望每次迭代使用每个核心的行块,因为一次将多行插入数据库更有效(如果使用典型的映射方法,则不是1乘1)
谢谢你的帮助.
最佳答案
不是预先创建生成器并将它们传递到每个线程,而是将其留给线程代码.
def processChunk(params):
filename,chunk_number,number_of_chunks = params
with open(filename,'r') as fp:
for line in file_block(fp,chunk_number):
process(line)
li = [(filename,i,number_of_chunks) for i in range(number_of_chunks)]
p.map(processChunk,li)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。