Read one million records per second

Question

The following code ingests 10k-20k records per second and I want to improve the performance of it. I'm reading json and ingesting it into the database using Kafka. I'm running it on the cluster of five nodes with zookeeper and Kafka installed on it.

Can you give me some tips to improve?

import os
import json
from multiprocessing import Pool
from kafka.client import KafkaClient
from kafka.producer import SimpleProducer


def process_line(line):
    producer = SimpleProducer(client)
    try:
       jrec = json.loads(line.strip())
       producer.send_messages('twitter2613',json.dumps(jrec))
    except ValueError, e:
                {}


if __name__ == "__main__":
    client = KafkaClient('10.62.84.35:9092')
    myloop=True
    pool = Pool(30)


    direcToData = os.listdir("/FullData/RowData")
    for loop in direcToData:
        mydir2=os.listdir("/FullData/RowData/"+loop)

        for i in mydir2:
            if  myloop:
                 with open("/FullData/RowData/"+loop+"/"+i) as source_file:
                     # chunk the work into batches of 4 lines at a time
                     results = pool.map(process_line, source_file, 30)

Answer 1

Coding style

1. get consistent: most of your functions and variable names are lower_snake_case except direcToData (why not directToData?), you should stick to lower_snake_case which is recommended by PEP8.

• get consistent: you should have spacing around all your assignments (=).
• use except ValueError as e:, the form using a coma is being deprecated and a syntax error in Python 3;
• use pass instead of {} to create a no-op statement;
• try to be consistent: indent with 4 spaces anywhere, not 9 as in your except;
• myloop (why not my_loop?) is set to True and never changed, thus if myloop is unnecessary;
• two blank lines better separate functions and classes visually. But within the same block it feels weird, you should remove one between pool = and direcToData =.

Oh, and you use both ' and " as string delimiter. You should try to use only one for consistency.

Directory traversal

You may be able to simplify your file management using os.walk. If "/FullData/RowData" recursively contains only the files you wish to process, then you can write:

for path, dirnames, filenames in os.walk('/FullData/RowData'):
    if filenames: # files exist in traversed directory
        full_names = map(lambda f, path=path: os.path.join(path, f), filenames)
        pool.map(process_file, full_names, 16) # Using @holroy advices

The process_file function would then be:

def process_file(filename):
    producer = SimpleProducer(client)
    with open(filename) as source_file:
        for line in source_file:
            try:
                jrec = json.loads(line.strip())
                producer.send_messages('twitter2613', json.dumps(jrec))
            except ValueError: # `as` nothing since we don't use the value
                pass

This will also help process things faster since it will create only one producer per file instead of one per line.

Answer 2

Don't split reading from same file into different processes

I haven't used Pool my self, but it seems like you're a splitting the file read over 30 different processes, where each reads 30 lines each? If that is correct, you should seriously consider a different split tactic, as that will throttle your IO seriously. You'll have 30 different processess trying to read from 30 different places in the file at the same time.

A better tactic would be to send each file to a different process, and then let that process handle that file completely.

Choose number of processes wisely

Another caveat would be the number of processes you use. You create 30 processes, but as long as you don't have a 30 actual processors available you wont see any major performance gain using this number of processes.

Back in the days using various Unix based operating systems, we did compilations in batches and the general rule was that we would aim for approx 4 times the number of processors we had available. In other words on a quad-processor, we would aim for 16 processes. Any more and we started seeing congestion due to interprocess issues and IO related performance bottlenecks.

Shorten the distance from file to server

Another speedup can be found if you are able to avoid network traffic. That is if you are able to run this script directly on the Kafka server, so that you can use loopback addresses and local connections instead of using the IP network.

Establish a baseline

Not so much a performance suggestion, but do you have good baselines for how long it takes to do a typical run using only a single process? This can be helpful, when you start dividing the load according to other metrics, to compare when you reach a threshold regarding what each server/client should do.

Answer 3

You have strange error handling in here:

except ValueError, e:
            {}

Why do you create a redundant dictionary? If you want nothing to happen, you can use the pass keyword. It's designed to be placed where Python expects blocks of code so it can just do nothing. This avoids the syntax error in trying to leave out a block of code. It's much more readable to do that way than creating an empty dictionary for no reason.

If you do have a good reason, then leave a comment explaining it as it's not clear in the current context.

Also don't use except ValueError, e because it's ambiguous (and specifically invalid syntax in Python 3). Consider if you needed to also catch an IOError:

except ValueError, IOError, e:

Now how is Python supposed to parse that? The comma separation doesn't make it clear what is a new exception and what should be the exception's name, so in Python 2 you should use the more explicit except ValueError as e form, which is also more readable. Incidentally this makes it easier to now use multiple exceptions:

except (ValueError, IOError) as e:

The myloop value seems to always be True, nothing in this code ever changes it. So why is it even necessary?

You're also using nested loops to find all the file paths that are inside /FullData/RowData, but glob could do this in one function call.

from glob import iglob

files = iglob('/FullData/RowData/*/*')

(glob is the normal function that returns a list but iglob returns a generator, which suits your needs better as it's more efficient when you're just iterating over it)

The *s are wildcards, so any folder/file in those places will be matched. But you could make it more specific by adding say a filename at the end:

files = iglob('/FullData/RowData/*/*.data')

Now all .data files in the nested folders will match. You know your filenames better than me, so you could figure out a good pattern. This means that you having a single loop:

if __name__ == "__main__":
    client = KafkaClient('10.62.84.35:9092')
    myloop=True
    pool = Pool(30)


    for f in iglob('/FullData/RowData/*/*.data'):
         with open(f) as source_file:
             # chunk the work into batches of 4 lines at a time
             results = pool.map(process_line, source_file, 30)