forked pyserini Collection iterators sometimes freeze with large directories about pyserini HOT 8 CLOSED

from pyserini.collection import pycollection
from pyserini.index import pygenerator

collection = pycollection.Collection('TrecCollection', '/tuna1/collections/newswire/disk45/')
generator = pygenerator.Generator('JsoupGenerator')

segments = [fs for (i, fs) in enumerate(collection)]

def do_something(fs):
    for (j, doc) in enumerate(fs):
        parsed = generator.create_document(doc)
        docid = parsed.get('id')            # FIELD_ID
        raw = parsed.get('raw')             # FIELD_RAW
        contents = parsed.get('contents')   # FIELD_BODY
    print('Finished {}, {} docs'.format(doc, j))

import concurrent.futures
import time
import sys

tic = time.clock()
with concurrent.futures.ThreadPoolExecutor(max_workers=int(sys.argv[1])) as executor:
    executor.map(do_something, segments)
toc = time.clock()
print('{} threads, total time: {} seconds'.format(sys.argv[1], toc-tic))

from pyserini.collection import pycollection
from pyserini.index import pygenerator

collection = pycollection.Collection('HtmlCollection', 'collection/')
generator = pygenerator.Generator('JsoupGenerator')

segments = [fs for (i, fs) in enumerate(collection)]

def do_something(fs):
    for (j, doc) in enumerate(fs):
        parsed = generator.create_document(doc)
        docid = parsed.get('id')            # FIELD_ID
        raw = parsed.get('raw')             # FIELD_RAW
        contents = parsed.get('contents')   # FIELD_BODY
        print('{} {} {}...'.format(j, docid, contents.strip().replace('\n', ' ')[:50]))

from multiprocessing import Pool

p = Pool(3)
p.map(do_something, segments)

from pyserini.collection import pycollection
from pyserini.index import pygenerator

collection = pycollection.Collection('HtmlCollection', 'collection/')

def do_something(fs):
    generator = pygenerator.Generator('JsoupGenerator')
    for (j, doc) in enumerate(fs):
        parsed = generator.create_document(doc)
        docid = parsed.get('id')            # FIELD_ID
        raw = parsed.get('raw')             # FIELD_RAW
        contents = parsed.get('contents')   # FIELD_BODY
        print('{} {} {}...'.format(j, docid, contents.strip().replace('\n', ' ')[:50]))

from multiprocessing import Process

for (i, fs) in enumerate(collection):
    proc = Process(target=do_something, args=(fs,))
    proc.start()

import os
from multiprocessing import Manager, Process, Pool
import time

def spawn_child_process_to_read_docs(data):

    from pyserini.index import pygenerator
    from pyserini.collection import pycollection
    path = data["rootdir"]
    ctype = data["ctype"]

    start = time.time()
    collection = pycollection.Collection(ctype, path)
    local_dict = {}
    generator = pygenerator.Generator("JsoupGenerator")
    for i, file_segment in enumerate(collection):
        doc_ids, doc_contents = read_file_segment(file_segment, generator)
        for i, doc_id in enumerate(doc_ids):
            local_dict[doc_id] = doc_contents[i]
 


def read_file_segment(file_segment, generator):
    doc_ids = []
    docs = []
    for j, doc in enumerate(file_segment):
        if doc.id is not None:
            parsed = generator.create_document(doc)
            if parsed is None:
                continue
            current_doc_id = parsed.get("id")
            contents = parsed.get("contents")    
            doc_ids.append(current_doc_id)
            docs.append(contents)

    return doc_ids, docs

if __name__ == "__main__":
    ctype = "TrecCollection"
    path = "/GW/NeuralIR/nobackup/Aquaint-TREC-3-4/NEWS_data"
    start_time = time.time()

    args_list = []
    for subdir in os.listdir(path):
        if os.path.isdir(path + "/" + subdir):
            args_list.append({"rootdir": path + "/" + subdir, "ctype": ctype})


    pool = Pool(processes=8)
    pool.map(spawn_child_process_to_read_docs, args_list)
    print("Getting all documents from disk took: {0}".format(time.time() - start_time))

from pyserini.collection import pycollection
from pyserini.index import pygenerator

collection = pycollection.Collection('TrecCollection', '/tuna1/collections/newswire/disk45/')
paths = [path.toString() for path in collection.get_segment_paths().toArray()]

def do_something(p):
    segment_raw = collection.create_file_segment(pycollection.JPaths.get(p))
    iter = segment_raw.iterator()
    print(f'segment: {p}')

    while iter.hasNext():
        doc = iter.next()

import concurrent.futures
import time
import sys

tic = time.clock()
with concurrent.futures.ThreadPoolExecutor(max_workers=int(sys.argv[1])) as executor:
    executor.map(do_something, paths)
toc = time.clock()
print('{} threads, total time: {} seconds'.format(sys.argv[1], toc-tic))