Мне нужно посчитать частоту слов в 3 ГБ сжатом текстовом файле с английскими предложениями, что составляет около 30 ГБ в разархивированном виде.
У меня есть однопоточный скрипт с collections.Counter и gzip.open, этоНа завершение работы уходит несколько часов.
Поскольку чтение файла строка за строкой происходит намного быстрее, чем разделение и подсчет, я думаю о потоке производителя-потребителя с устройством чтения файлов для создания строк и нескольких потребителей для выполнения разделения исчитая, и в конце, объедините Counter s, чтобы получить вхождение слова.
Однако я не могу найти пример для ProcessPoolExecutor, чтобы отправить очередь на Executor, они просто mapодин элемент из списка.Для asyncio.Queue.
есть только однопоточные примеры. Это огромный файл, поэтому я не могу прочитать весь файл и получить list до подсчета, поэтому я не могу использовать concurrent.futures.Executor.map.Но во всех примерах, которые я читаю, в качестве начала используется фиксированный список.
Время разделения и подсчета одного предложения сравнимо с разветвлением процесса, поэтому я должен сделать каждый потребительский процесс более продолжительным.Я не думаю, что map может объединить Counter s, поэтому я не могу использовать 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?