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) except: print('Index out of bound error due to NULL Values') continue for keyword in range(0,len(keywords)): outFile.write(keywords[keyword] + '\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.