Extracting keywords from 3 billion CSV records

Question

I am processing around 3 billion records using this piece of code. It's pretty slow; it would be really helpful if you could suggest better ways to do this.

'''
Created on 27-Mar-2015
@author: siddarth
'''

import rake
import multiprocessing as mp
import time
import os
import csv

filename = 'Large_Input_File.csv'
outName = 'Large_Output_File.csv'
BYTES_PER_MB = 1048576

start = time.time()
def elapsed():
    return time.time() - start


''' Worker function used to call rake 
    This method basically calls the rake class which returns a 
    list of keywords, also each process 
    writes to a different output file'''

def fileparser_worker(filename, start, end, c):
    outFileName = outName + str(c) 
    outFile = open(outFileName,'w')
    with open(filename) as inFile:
        inFile.seek(start)
        lines = inFile.readlines(end-start)
        for line in lines:
            title = line.split('\t') 
            try:
                keywords = rake.execute_rake(title[2])
            except:
                print('Index out of bound error due to NULL Values')
                continue
            for keyword in range(0,len(keywords)):
                outFile.write(keywords[keyword][0] + '\n')
    outFile.close()

if __name__ == '__main__':
    start = time.time()
    chunk_start = 0
    chunk_size =  512 * BYTES_PER_MB
    chunk_end = 512 * BYTES_PER_MB

    filesize = os.path.getsize(filename)

    print '\n%.3fs: file has %s rows' % (elapsed(), filesize)

    c = 0
    pause = 0
    iterations = (filesize / chunk_size) + 1
    ''' Chunk the file ''' 
    print(iterations)
    with open(filename) as inFile:
        while c < iterations:
            if chunk_start + chunk_size > filesize:
                chunk_end = filesize
            else:
                chunk_end = chunk_start + chunk_size

            inFile.seek(chunk_end)
            line = inFile.readline()

            if line == '':
                continue
            else:
                chunk_end = inFile.tell()

            print("Start chunk",chunk_start)
            ''' Initializing Processes '''
            proc = mp.Process(target=fileparser_worker, args = (filename,chunk_start,chunk_end, c) )
            proc.start()
            pause += 1
            print("End chunk",chunk_end)
            chunk_start = chunk_end
            c += 1

            ''' Making sure that there are atmost 4 processes 
                running at a single point of time,
                cause I have a 4 core machine'''

            if pause == 4:
                proc.join() # Waits for processes to close 
                pause = 0

    proc.join() # Close off any running processes
    print("TOTAL TIME TAKEN",elapsed())

Below are some metrics which may help you to understand what's going on under the hood.

File Size = 18GB

Usage Metrics:
CPU Usage = 99% ( Using 4 Processes )
RAM Usage = 30% ( May be I can use more RAM to improve performance ? )

I am not familiar with multiprocessing. I've learned about it today and wrote some sample code based on what I understood.

Answer 1

def fileparser_worker(filename, start, end, c):
    with open(filename) as inFile, open(outName + str(c),'w') as outFile:
            inFile.seek(start)
            #lines = inFile.readlines(end-start)

because readlines calls readline multiple times it can be replaced with read.

If the text should be split by newline and each split line should be splitted by a tab character, there are two ways to do it depending on the input text.

    for title in (line.split('\t') for line inFile.read(end-start).splitlines()):  

Here title is list of strings.

Another one is to call str.split() or str.split(None) in which case the string is split by white spaces (space, tab or newline), this is best if it's ok to split by space character or the text has no space character.

    for title in inFile.read(end-start).split():  

Here the title is string.

If the read().split() becomes a memory hog, I think a generator will give better result.

    #def get_title():
    #   yield next((line.split('\t') for line inFile.read(end-start).splitlines()))

    for title in (line.split('\t') for line inFile.read(end-start).splitlines()):  
        try:
            keywords = rake.execute_rake(title[2])
        except:
            print('Index out of bound error due to NULL Values')
            continue

If we could build a sequence of strings out of keywords we can call writelines straight.

        else:
            outFile.writelines((keyword[0] + '\n' for keyword in keywords))

~

Answer 2

Motherboards with multiple processors, processors with multiple cores, and the processor's microarchitecture will greatly affect the benefits of parallel processing, and multithreading.

Without knowing your motherboard and which processors are used, I am very limited as to what techniques will help or penalize your approach.

It appears you have not implemented any thread synchronization, communications or division of tasks for optimized allocation of resources.

There is not much chance you can take advantage of multithreading your app. Unless your workstation uses an Intel micro architecture such as Haswell where every core has its own L1 cache, the motherboard has dual memory access, and the PCIe bus is not shared with the memory bus.

In a nutshell, you have to split up the tasks. Running the same code on multiple threads is not going to help much. Your task list is bringing in data from the hard drive (SSD would help, or RAM disk), and store it in memory, search the data, store the results. Given your system architecture if any of these tasks can run simultaneously you can allocate those tasks to separate threads.

Basically you do not want multiple threads accessing the same cache memory. Set up your memory to best utilize the size of your on chip cache. You would have to coordinate the resources between the threads with cooperative multitasking. Set up a queue between the processes that coordinates the threads to prevent accessing the same resources at the same time.

I doubt you can get a big boost from multitasking. Your best chance for performance gains is in optimizing your search.

Just because your CPU is at 99% does not mean each core is crunching your code. Most likely they are sitting in wait loops waiting for access to resources.

With 4 cores you should likely be running 3 or fewer threads. Your code be penalized for your uncooperative method. You likely have a lot of cache eviction going on.

If you are going to use multiple threads, put in pauses to allow the other threads to get out of their wait loop. Keep loop iterations small.

You biggest problem is an 18GB flat file with no indexing.

You want to have the data stored in a way where as much of the processing is done in the data structure and organization.

Get rid of the CSV. The only way to increase performance would be to break up the job to run on separate workstations.

Look into how search engines store the key words. There are some open source projects that work fairly well. Consider using Google's search services if you don't care they are going to analyze your data to find a way to capitalize on it. If you were to use 3rd party resources READ THE TERMS OF SERVICE! It is absolutely amazing how little people know what Google is doing. More and more are learning from Google and doing the same appalling stuff.

A project like Heritrix, Solr, or even Sphider will do much better than an un-indexed CSV flat file.

Break the file into pieces and store them in pieces. But only if you have a legitimate reason for keeping the CSV files. Why use your code to do that task when it can be done off line.

Your best bet is to index the data.

In PHP I use a text file technique that beats CSV. I organize the data into huge arrays serialize them and save them as text files. Convert your CSV into arrays, serialize then, gzip them and store.

Size the arrays to fit in L1 or L2 cache. Pad them if necessary to organize then on memory page boundaries.

If you have not read Intel's 64 and IA-32 Architectures Optimization Reference Manual, read it. All programmers should be familiar with it.

You don't have to read all 642 pages. At a minimum read Chapter 3 and read the User/Source Coding Rules, search the document using this search phrase: "User/Source Coding Rule". Lots of tips for multithreading.

The most important: Source coding rules pertaining to branching (if else), loops (while), and variable declaration and organization.

My number one programming rule for fast code: Use arrays to eliminate branches in your code (see Intel's Optimization Manual).

This is assuming that retrieval of comments and generation of the HTML takes about 50-100 micro-seconds per comment is an improvement over Disqus. The number of comments in the database has no affect on performance. The query is a single simple, query for all the records (including threading) associated with the topic.