如何同时计算大文件中的词频？

Question

我需要计算一个3GB的gzip纯文本英文句子文件的词频，解压时大约30GB。我有一个带有collections.Counterand 的单线程脚本gzip.open，需要几个小时才能完成。由于逐行读取文件比拆分和计数快得多，因此我正在考虑使用文件读取器生成行和多个消费者进行拆分和计数的生产者-消费者流程，最后，将Counters合并到获取单词出现。但是，我找不到ProcessPoolExecutor将队列发送到的示例Executor，它们只是map列表中的单个项目。. 只有单线程示例asyncio.Queue。这是一个巨大的文件，所以我无法读取整个文件并获得list之前的计数，因此我无法使用concurrent.futures.Executor.map. 但是我阅读的所有示例都使用固定列表作为开始。拆分和计算一个句子的时间相当于fork一个进程，所以我必须让每个消费者进程的寿命更长。我不认为mapcan merge Counters，所以我不能使用chunksize> 1。因此我必须给消费者一个队列并让他们继续计数直到整个文件完成。但大多数示例只向消费者发送一件物品并使用chunksize=1000以减少fork时间。你能帮我写一个例子吗？我希望代码向后兼容 Python 3.5.3，因为 PyPy 更快。我的真实案例是针对更具体的文件格式：chr1    10011   141     0       157     4       41      50chr1    10012   146     1       158     4       42      51chr1    10013   150     0       163     4       43      53chr1    10014   164     3       167     4       44      54我需要计算第 3 到第 8 列的单列的每个直方图。所以我将词频作为一个更简单的例子。csv.DictReader 花费大部分时间。我的问题是，虽然 gzip 阅读器很快，但 csv 阅读器很快，我需要计算数十亿行。而且 csv 阅读器肯定比 gzip 阅读器慢。因此，我需要将行传播到 csv 读取器的不同工作进程并分别进行下游计数。在一个生产者和许多消费者之间使用队列很方便。由于我使用的是 Python，而不是 C，是否有一些抽象的多处理和队列包装器？这可以ProcessPoolExecutor与Queue类一起使用吗？

Answer 1

一个 30 GB 的文本文件足以将您的问题放入大数据领域。所以为了解决这个问题，我建议使用大数据工具，比如 Hadoop 和 Spark。您所解释的“生产者-消费者流”基本上就是MapReduce算法的设计目的。字数频率是典型的 MapReduce 问题。查一查，你会发现很多例子。

Answer 2

我在周末学习了多处理库。

停止按 Ctrl+C 并写入当前结果功能仍然无效。

主要功能现在很好。

#!/usr/bin/env pypy3

import sys

from collections import Counter

from multiprocessing import Pool, Process, Manager, current_process, freeze_support

SamplesList = ('D_Crick', 'D_Watson', 'Normal_Crick', 'Normal_Watson', 'D_WGS', 'Normal_WGS')

ChunkSize = 1024 * 128

verbose = 0

Nworkers = 16

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 = CallStat(inDepthFile)

    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(DepthCnt[col][depth]) for col in SamplesList)) )

        #print( '{}\t{}'.format(depth,'\t'.join(str(yDepthCnt[depth][col]) for col in SamplesList)) )

        print( '{}\t{}'.format(depth,'\t'.join(str(cDepthCnt[col][depth]) for col in SamplesList)),file=tsvout)

        #pass

    #print('#MaxDepth={}'.format(MaxDepth),file=tsvout)

    tsvout.close()

    pass

def CallStat(inDepthFile):

    import gzip

    import itertools

    RecordCnt = 0

    MaxDepth = 0

    cDepthCnt = {key:Counter() for key in SamplesList}

    cDepthStat = {key:[0,0,0,0,0] for key in SamplesList} # x and x^2

    #lines_queue = Queue()

    manager = Manager()

    lines_queue = manager.Queue()

    stater_pool = Pool(Nworkers)

    TASKS = itertools.repeat((lines_queue,SamplesList),Nworkers)

    #ApplyResult = [stater_pool.apply_async(iStator,x) for x in TASKS]

    #MapResult = stater_pool.map_async(iStator,TASKS,1)

    AsyncResult = stater_pool.imap_unordered(iStator,TASKS,1)

    try:

        with gzip.open(inDepthFile, 'rt') as tsvfin:

            while True:

                lines = tsvfin.readlines(ChunkSize)

                lines_queue.put(lines)

                if not lines:

                    for i in range(Nworkers):

                        lines_queue.put(b'\n\n')

                    break

    except KeyboardInterrupt:

        print('\n[!]Ctrl+C pressed.',file=sys.stderr,flush=True)

        for i in range(Nworkers):

            lines_queue.put(b'\n\n')

        pass

    #for results in ApplyResult:

        #(iRecordCnt,iMaxDepth,icDepthCnt,icDepthStat) = results.get()

    #for (iRecordCnt,iMaxDepth,icDepthCnt,icDepthStat) in MapResult.get():

    for (iRecordCnt,iMaxDepth,icDepthCnt,icDepthStat) in AsyncResult:

        RecordCnt += iRecordCnt

        if iMaxDepth > MaxDepth:

            MaxDepth = iMaxDepth

        for k in SamplesList:

            cDepthCnt[k].update(icDepthCnt[k])

            cDepthStat[k][0] += icDepthStat[k][0]

            cDepthStat[k][1] += icDepthStat[k][1]

    return RecordCnt,MaxDepth,cDepthCnt,cDepthStat

#def iStator(inQueue,inSamplesList):

def iStator(args):

    (inQueue,inSamplesList) = args

    import csv

    # 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 inSamplesList}

    cDepthStat = {key:[0,0] for key in inSamplesList} # x and x^2

    RecordCnt = 0

    MaxDepth = 0

    for lines in iter(inQueue.get, b'\n\n'):

        try:

            tsvin = csv.DictReader(lines, delimiter='\t', fieldnames=('ChrID','Pos')+inSamplesList )

            for row in tsvin:

                #print(', '.join(row[col] for col in inSamplesList))

                RecordCnt += 1

                for k in inSamplesList:

                    theValue = int(row[k])

                    if theValue > MaxDepth:

                        MaxDepth = theValue

                    #DepthCnt[k][theValue] += 1  # PyPy3:30.54 ns, Python3:22.23 ns

                    #yDepthCnt[theValue][k] += 1 # PyPy3:30.47 ns, Python3:21.50 ns

                    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(current_process().name,RecordCnt,MaxDepth),file=sys.stderr,flush=True)

    return RecordCnt,MaxDepth,cDepthCnt,cDepthStat

if __name__ == "__main__":

    main()  # time python3 ./samdepthplot.py t.tsv.gz 1

Answer 3

只是一些伪代码：

from concurrent.futures import ProcessPoolExecutor

from multiprocessing import Manager

import traceback

WORKER_POOL_SIZE = 10  # you should set this as the number of your processes

QUEUE_SIZE = 100       # 10 times to your pool size is good enough

def main():

    with Manager() as manager:

        q = manager.Queue(QUEUE_SIZE)

        # init worker pool

        executor = ProcessPoolExecutor(max_workers=WORKER_POOL_SIZE)

        workers_pool = [executor.submit(worker, i, q) for i in range(WORKER_POOL_SIZE)]

        # start producer

        run_producer(q)

        # wait to done

        for f in workers_pool:

            try:

                f.result()

            except Exception:

                traceback.print_exc()

def run_producer(q):

    try:

        with open("your file path") as fp:

            for line in fp:

                q.put(line)

    except Exception:

        traceback.print_exc()

    finally:

        q.put(None)

def worker(i, q):

    while 1:

        line = q.get()

        if line is None:

            print(f'worker {i} is done')

            q.put(None)

            return

        # do something with this line

        # ...