【发布时间】：2019-07-31 01:31:58
【问题描述】：
我需要统计一个3GB的英文句子gzip压缩纯文本文件的词频，解压后大约30GB。
我有一个带有collections.Counter 和gzip.open 的单线程脚本，需要几个小时才能完成。
由于逐行读取文件比拆分和计数快得多，因此我正在考虑使用文件读取器生成行和多个消费者进行拆分和计数的生产者-消费者流，最后合并Counters 获取单词的出现次数。
但是，我找不到 ProcessPoolExecutor 将队列发送到 Executor 的示例，它们只是列表中的 map 单个项目。
asyncio.Queue 只有单线程示例。
这是一个巨大的文件，所以我无法读取整个文件并在计数之前获得
list，因此我无法使用concurrent.futures.Executor.map。但是我阅读的所有示例都使用固定列表作为开始。拆分和计数一个句子的时间相当于fork一个进程，所以我必须让每个消费者进程的寿命更长。我不认为
map可以合并Counters，所以我不能使用chunksize>1。因此，我必须给消费者一个队列，让他们继续计数，直到整个文件完成。但大多数示例只向消费者发送一件商品，并使用chunksize=1000减少fork次。
你能帮我写一个例子吗？
我希望代码向后兼容 Python 3.5.3，因为 PyPy 更快。
我的真实情况是针对更具体的文件格式：
chr1 10011 141 0 157 4 41 50
chr1 10012 146 1 158 4 42 51
chr1 10013 150 0 163 4 43 53
chr1 10014 164 3 167 4 44 54
我需要计算从第 3 列到第 8 列的单列的每个直方图。 所以我以词频为例。
我的代码是：
#!/usr/bin/env pypy3
import sys
SamplesList = ('D_Crick', 'D_Watson', 'Normal_Crick', 'Normal_Watson', 'D_WGS', 'Normal_WGS')
def main():
import math
if len(sys.argv) < 3 :
print('Usage:',sys.argv[0],'<samtools.depth.gz> <out.tsv> [verbose=0]',file=sys.stderr,flush=True)
exit(0)
try:
verbose = int(sys.argv[3])
except: # `except IndexError:` and `except ValueError:`
verbose = 0
inDepthFile = sys.argv[1]
outFile = sys.argv[2]
print('From:[{}], To:[{}].\nVerbose: [{}].'.format(inDepthFile,outFile,verbose),file=sys.stderr,flush=True)
RecordCnt,MaxDepth,cDepthCnt,cDepthStat = inStat(inDepthFile,verbose)
for k in SamplesList:
cDepthStat[k][2] = cDepthStat[k][0] / RecordCnt # E(X)
cDepthStat[k][3] = cDepthStat[k][1] / RecordCnt # E(X^2)
cDepthStat[k][4] = math.sqrt(cDepthStat[k][3] - cDepthStat[k][2]*cDepthStat[k][2]) # E(X^2)-E(X)^2
tsvout = open(outFile, 'wt')
print('#{}\t{}'.format('Depth','\t'.join(SamplesList)),file=tsvout)
#RecordCntLength = len(str(RecordCnt))
print( '#N={},SD:\t{}'.format(RecordCnt,'\t'.join(str(round(cDepthStat[col][4],1)) for col in SamplesList)),file=tsvout)
for depth in range(0,MaxDepth+1):
print( '{}\t{}'.format(depth,'\t'.join(str(cDepthCnt[col][depth]) for col in SamplesList)),file=tsvout)
tsvout.close()
pass
def inStat(inDepthFile,verbose):
import gzip
import csv
from collections import Counter
# Looking up things in global scope takes longer then looking up stuff in local scope. <https://stackoverflow.com/a/54645851/159695>
cDepthCnt = {key:Counter() for key in SamplesList}
cDepthStat = {key:[0,0,0,0,0] for key in SamplesList} # x and x^2
RecordCnt = 0
MaxDepth = 0
with gzip.open(inDepthFile, 'rt') as tsvin:
tsvin = csv.DictReader(tsvin, delimiter='\t', fieldnames=('ChrID','Pos')+SamplesList )
try:
for row in tsvin:
RecordCnt += 1
for k in SamplesList:
theValue = int(row[k])
if theValue > MaxDepth:
MaxDepth = theValue
cDepthCnt[k][theValue] += 1 # PyPy3:29.82 ns, Python3:30.61 ns
cDepthStat[k][0] += theValue
cDepthStat[k][1] += theValue * theValue
#print(MaxDepth,DepthCnt)
except KeyboardInterrupt:
print('\n[!]Ctrl+C pressed.',file=sys.stderr,flush=True)
pass
print('[!]Lines Read:[{}], MaxDepth is [{}].'.format(RecordCnt,MaxDepth),file=sys.stderr,flush=True)
return RecordCnt,MaxDepth,cDepthCnt,cDepthStat
if __name__ == "__main__":
main() # time python3 ./samdepthplot.py t.tsv.gz 1
csv.DictReader 花费的时间最多。
我的问题是，虽然 gzip 阅读器速度很快，但 csv 阅读器速度很快，但我需要计算数十亿行。而且 csv 阅读器肯定比 gzip 阅读器慢。
所以，我需要将行分散到 csv 阅读器的不同工作进程中，并分别进行下游计数。在一个生产者和多个消费者之间使用队列很方便。
由于我使用的是 Python，而不是 C，是否有一些用于多处理和队列的抽象包装器？这可以将ProcessPoolExecutor 与Queue 类一起使用吗？
【问题讨论】：
-
我知道你只是简单地提到了你是如何做到这一点的，但是你能把你现在使用的代码包括进来吗？
-
我想知道您是否仅使用 shell 管道可能无法获得更好的性能？见Command-line Tools can be 235x Faster than your Hadoop Cluster。这个问题听起来很适合
xargs和uniq -c，也许有一些awk脚本将它们粘合在一起。 -
您是否考虑过使用
io.BufferedReader？如Reading & Writing GZIP Files Faster in Python 中所述 -
您可以将 gzip 压缩文件视为一个巨大的随机访问行列表，而无需使用类似于 answer 中正在执行的操作将整个内容读入内存，仅使用
mmap而不是一个临时文件（我有一个未发布的版本）。然后可以将内存映射连同起始行号和行数一起传递给多个并发子进程。每个子进程可以计算分配给它的部分中的单词，并在完成时传回字典。这些字典可以全部合并在一起。 -
我现在包含了我的代码。
标签： python parallel-processing queue process-pool