概述
我目前正在尝试读取一个大文件(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)
总结
以上是编程之家为你收集整理的Python：使用多个核心的流程文件全部内容，希望文章能够帮你解决Python：使用多个核心的流程文件所遇到的程序开发问题。
如果您也喜欢它,动动您的小指点个赞吧
602392714
清零编程群