5
\$\begingroup\$

I'm trying to process a large dataset (300GB, myfile.txt in the script) line by line using multiprocessing. I want to define a 1% random sample based one variable (contained in unique_ids_final.txt). My first step is to define the sample and then I want to read the data file using multiprocessing. I would like to improve the efficiency of the code in two ways:

First, I'd like to pass the pct1 object from the parent to the child so that it only needs to be defined once. I've seen a description of this on thelaziestprogrammer.com: Pass Data to Workers w/o Globals, but I'm relatively new to python and don't understand how to implement it in my code.

Second, I'd like to define the chunks without reading in the data. In chunkify() I get the start of each chunk and the number of bytes that need to be read by looking for the end of the line after reading in 1MB of data. I was hoping to use seek to move forward by 1MB and then find the end of the line, but this creates problems because later I need to read in the chunks and read treats '\n' as one byte, while seek treats it as two.

Any other suggestions to increase efficiency would also be much appreciated!

#define sample
uid = list(line.strip() for line in open('Subsets/unique_ids_final.txt'))
pct1 = round(len(uid)/100)
random.seed(1)
id_pct1 = set(random.sample(uid, k=pct1))
id_pct1.add(vname)

#read original file and write 1% sample using multiprocessing
def worker(chunkStart, chunkSize, q):
    with open('myfile.txt') as f:
        tlines = []
        f.seek(chunkStart)
        lines = f.read(chunkSize).splitlines()
        for line in lines:
            data = line.split('*')
            if data[30] in id_pct1: tlines.append(line)
        q.put(tlines)
        return tlines

def chunkify(fname,size=1024*1024):
    fileEnd = os.path.getsize(fname)
    with open(fname, 'r') as f:
        chunkEnd2 = 0
        while True:
            chunkStart = chunkEnd2
            f.seek(chunkStart)
            f.read(size)
            chunkEnd1 = f.tell()
            f.readline()
            chunkEnd2 = f.tell()
            chunkSz = 1024*1024 + chunkEnd2 - chunkEnd1 - 1
            yield chunkStart, chunkSz
            if chunkEnd2 >= fileEnd:
                break

def listener(q):
    with open('myfile1pct.txt', 'w') as out_f1:
        while True:
            m = q.get()
            if m == 'kill': break
            else:
                for line in m:
                    out_f1.write(line+'\n')
                    out_f1.flush()

def main():

    manager = mp.Manager()
    q = manager.Queue()
    pool = mp.Pool()

    watcher = pool.apply_async(listener, (q,))

    jobs = []
    for chunkStart, chunkSize in chunkify('myfile.txt'):
        jobs.append(pool.apply_async(worker,(chunkStart,chunkSize,q)))

    for job in jobs:
        job.get()

    q.put('kill')
    pool.close()
    pool.join()

if __name__ == '__main__':
    main()
\$\endgroup\$
  • \$\begingroup\$ Can you share some information about the data itself? Ideally we would have enough to run the program, since matters of performance are so dependent on benchmarking and profiling. \$\endgroup\$ – AMC Dec 23 '19 at 1:54
  • \$\begingroup\$ It is insurance claims data, which is privacy protected so I don't know of any sample data that's out there. There are ~300 million lines in the file. Each line represents a claim line and has 171 variables that are delimited with *. I make the 1% sample at the person level, using a list of 4 million person ids represented by integers and contained in idunique_ids_final. Let me know if there's any other useful information I can share. \$\endgroup\$ – giacomo1488 Dec 24 '19 at 15:20
  • \$\begingroup\$ The 4 million individual IDs are used to determine which claims to extract? \$\endgroup\$ – AMC Dec 24 '19 at 16:00
  • \$\begingroup\$ Yes, a 1% sample of the 4 million IDs. So I'm extracting the claims for 40,000 people. \$\endgroup\$ – giacomo1488 Dec 26 '19 at 14:50
  • \$\begingroup\$ I somehow forgot about this question, but I will return to it... \$\endgroup\$ – AMC Dec 30 '19 at 2:26
2
\$\begingroup\$

Disclaimer: I have never worked with multiprocessing, so I can't comment on that implementation.


Handling Files

I noticed at the top of the file you open a file, but never close it. (For the argument that an anonymous file is closed right after the statement is executed, see this post). Leaving files open is not a good idea. You should always close your files; leaving files open can slow down your program. I'd change that line to the code below:

with open('Subsets/unique_ids_final.txt') as file:
    uid = list(line.strip() for line in file.readlines())

While it is more typing, with automatically closes the file after the inner code is run.

List Comprehension

A couple of you for loops can be reduced to one line statements. You can create a list directly with a for loop. Take a look:

tlines = [line for line in lines if line.split('*')[30] in id_pct1]

Same with jobs:

jobs = [
    pool.apply_async(worker, (chunk_start, chunk_size, q))
    for chunk_start, chunk_size in chunkify('myfile.txt')
]

One line if statements

This

if m == 'kill': break

should be this

if m == 'kill':
    break

Even though it's one line, one word, you should still indent.

Docstrings

You should include docstrings when you write functions, methods, and classes. They are used to provide more description. Take a look:

def worker(chunk_start, chunk_size, q) -> List[str]:
    """
    Read original file and write 1% sample using multiprocessing

    :param <type> chunk_start: <description>
    :param <type> chunk_size: <description>
    :param <type> q: <description>

    :return: List[str]
    """

I had trouble following your code and understanding what variables were what types. Essentially, when labeling parameters in your docstrings, the format I used goes as follows:

:param <type of parameter> <name of parameter>: <description about parameter>

And returns are as follows:

:return: <type to return>

If you want docbuilders, consider using sphinx.

Type Hints

These help portray what types are accepted and returned by a function/method. Take a look:

def add(x: int, y: int) -> int:
    return x + y

While this is a very straightforward example, the idea is still there.

Variable/Parameter Names

These should be in snake_case, not mixedCase.

chunkStart -> chunk_start
chunkSize -> chunk_size
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thanks for your suggestions! \$\endgroup\$ – giacomo1488 Feb 10 at 17:25

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.