Processing huge CSV file using Python and multithreading

Question

I have a function that yields lines from a huge CSV file lazily:

def get_next_line():
    with open(sample_csv,'r') as f:
        for line in f:
            yield line

def do_long_operation(row):
    print('Do some operation that takes a long time')

I need to use threads such that each record I get from the above function I can call do_long_operation.

Most places on Internet have examples like this, and I am not very sure if I am on the right path.

import threading
thread_list = []
for i in range(8):
   t = threading.Thread(target=do_long_operation, args=(get_next_row from get_next_line))
   thread_list.append(t)

for thread in thread_list:
    thread.start()

for thread in thread_list:
    thread.join()

My questions are:

How do I start only a finite number of threads, say 8?
How do I make sure that each of the threads will get a row from get_next_line?

devilpreet · Accepted Answer · 2020-01-27 12:37:42Z

12

You could use a thread pool from multiprocessing and map your tasks to a pool of workers:

from multiprocessing.pool import ThreadPool as Pool
# from multiprocessing import Pool
from random import randint
from time import sleep


def process_line(l):
    print l, "started"
    sleep(randint(0, 3))
    print l, "done"


def get_next_line():
    with open("sample.csv", 'r') as f:
        for line in f:
            yield line

f = get_next_line()

t = Pool(processes=8)

for i in f:
    t.map(process_line, (i,))
t.close()
t.join()

This will create eight workers and submit your lines to them, one by one. As soon as a process is "free", it will be allocated a new task.

There is a commented out import statement, too. If you comment out the ThreadPool and import Pool from multiprocessing instead, you will get subprocesses instead of threads, which may be more efficient in your case.

edited Jan 27, 2020 at 12:37

devilpreet

7614 silver badges19 bronze badges

answered Jul 6, 2017 at 14:07

Hannu

12.3k4 gold badges38 silver badges52 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Hannu Over a year ago

I would use pool from multiprocessing instead. Give it a go and see which one works faster, but I am quite sure it will be the multiprocessing solution, but if you need to share a lot of internal variables or something like that, which is bad programming anyway, might prevent you from using multiprocessing pool. Function interface to both pools is exactly the same.

Peter Mortensen · Accepted Answer · 2019-06-03 10:53:11Z

2

Using a Pool/ThreadPool from multiprocessing to map tasks to a pool of workers and a Queue to control how many tasks are held in memory (so we don't read too far ahead into the huge CSV file if worker processes are slow):

from multiprocessing.pool import ThreadPool as Pool
# from multiprocessing import Pool
from random import randint
import time, os
from multiprocessing import Queue


def process_line(l):
    print("{} started".format(l))
    time.sleep(randint(0, 3))
    print("{} done".format(l))


def get_next_line():
    with open(sample_csv, 'r') as f:
        for line in f:
            yield line

# use for testing
# def get_next_line():
#     for i in range(100):
#         print('yielding {}'.format(i))
#         yield i


def worker_main(queue):
    print("{} working".format(os.getpid()))
    while True:
        # Get item from queue, block until one is available
        item = queue.get(True)
        if item == None:
            # Shutdown this worker and requeue the item so other workers can shutdown as well
            queue.put(None)
            break
        else:
            # Process item
            process_line(item)
    print("{} done working".format(os.getpid()))


f = get_next_line()

# Use a multiprocessing queue with maxsize
q = Queue(maxsize=5)

# Start workers to process queue items
t = Pool(processes=8, initializer=worker_main, initargs=(q,))

# Enqueue items. This blocks if the queue is full.
for l in f:
    q.put(l)

# Enqueue the shutdown message (i.e. None)
q.put(None)

# We need to first close the pool before joining
t.close()
t.join()

edited Jun 3, 2019 at 10:53

Peter Mortensen

31.4k22 gold badges110 silver badges134 bronze badges

answered Oct 31, 2018 at 4:20

KB5

214 bronze badges

2 Comments

user8188120 Over a year ago

In this example, how would you return an array of processed line elements? For example, say you were returning the first element in each line, how would you append them to an empty list iteratively and then return the final list of values?

KB5 Over a year ago

@user8188120 you could use a second queue (output queue) that the worker threads would write processed elements to. Create a queue and pass it into worker_main using initargs and then in the worker_main append processed elements to the queue. You could have another pool that reads processed elements off the output queue or read them at the end (after t.join()).

Peter Mortensen · Accepted Answer · 2019-06-03 10:50:59Z

Hannu's answer is not the best method. I ran the code on a 100M rows CSV file. It took me forever to perform the operation.

However, prior to reading his answer, I had written the following code:

def call_processing_rows_pickably(row):
    process_row(row)

import csv
from multiprocessing import Pool
import time
import datetime

def process_row(row):
    row_to_be_printed = str(row)+str("hola!")
    print(row_to_be_printed)

class process_csv():

    def __init__(self, file_name):
        self.file_name = file_name

    def get_row_count(self):
        with open(self.file_name) as f:
            for i, l in enumerate(f):
                pass
        self.row_count = i

    def select_chunk_size(self):
        if(self.row_count>10000000):
            self.chunk_size = 100000
            return
        if(self.row_count>5000000):
            self.chunk_size = 50000
            return
        self.chunk_size = 10000
        return

    def process_rows(self):
        list_de_rows = []
        count = 0
        with open(self.file_name, 'rb') as file:
            reader = csv.reader(file)
            for row in reader:
                print(count+1)
                list_de_rows.append(row)
                if(len(list_de_rows) == self.chunk_size):
                    p.map(call_processing_rows_pickably, list_de_rows)
                    del list_de_rows[:]

    def start_process(self):
        self.get_row_count()
        self.select_chunk_size()
        self.process_rows()

initial = datetime.datetime.now()
p = Pool(4)
ob = process_csv("100M_primes.csv")
ob.start_process()
final = datetime.datetime.now()
print(final-initial)

This took 22 minutes. Obviously, I need to have more improvements. For example, the Fred library in R takes 10 minutes maximum to do this task.

The difference is: I am creating a chunk of 100k rows first, and then I pass it to a function which is mapped by threadpool(here, 4 threads).

I don't think this would work in python3 since csv.reader can only read from a file opened in 'r' mode

Collectives™ on Stack Overflow

Processing huge CSV file using Python and multithreading

3 Answers 3

1 Comment

2 Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

2 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related