This might be too late, but just for future users I'll post anyway. Another poster mentioned using multiprocessing. I can vouch for it and can go into more detail. We deal with files in the hundreds of MB/several GB every day using Python. So it's definitely up to the task. Some of files we deal with aren't CSVs, so the parsing can be fairly complex and take longer than the disk access. However, the methodology is the same no matter what file type.
You can process pieces of the large files concurrently. Here's pseudo code of how we do it:
import os, multiprocessing as mp
# process file function
def processfile(filename, start=0, stop=0):
if start == 0 and stop == 0:
... process entire file...
else:
with open(file, 'r') as fh:
fh.seek(start)
lines = fh.readlines(stop - start)
... process these lines ...
return results
if __name__ == "__main__":
# get file size and set chuck size
filesize = os.path.getsize(filename)
split_size = 100*1024*1024
# determine if it needs to be split
if filesize > split_size:
# create pool, initialize chunk start location (cursor)
pool = mp.Pool(cpu_count)
cursor = 0
results = []
with open(file, 'r') as fh:
# for every chunk in the file...
for chunk in xrange(filesize // split_size):
# determine where the chunk ends, is it the last one?
if cursor + split_size > filesize:
end = filesize
else:
end = cursor + split_size
# seek to end of chunk and read next line to ensure you
# pass entire lines to the processfile function
fh.seek(end)
fh.readline()
# get current file location
end = fh.tell()
# add chunk to process pool, save reference to get results
proc = pool.apply_async(processfile, args=[filename, cursor, end])
results.append(proc)
# setup next chunk
cursor = end
# close and wait for pool to finish
pool.close()
pool.join()
# iterate through results
for proc in results:
processfile_result = proc.get()
else:
...process normally...
Like I said, that's only pseudo code. It should get anyone started who needs to do something similar. I don't have the code in front of me, just doing it from memory.
But we got more than a 2x speed up from this on the first run without fine tuning it. You can fine tune the number of processes in the pool and how large the chunks are to get an even higher speed up depending on your setup. If you have multiple files as we do, create a pool to read several files in parallel. Just be careful no to overload the box with too many processes.
Note: You need to put it inside an "if main" block to ensure infinite processes aren't created.
Answer from max on Stack OverflowParallel processing of a large .csv file in Python - Stack Overflow
Python, process a large text file in parallel - Stack Overflow
multithreading - How to read / process large files in parallel with Python - Stack Overflow
Python - How to parallel consume and operate on files in a directory - Stack Overflow
This might be too late, but just for future users I'll post anyway. Another poster mentioned using multiprocessing. I can vouch for it and can go into more detail. We deal with files in the hundreds of MB/several GB every day using Python. So it's definitely up to the task. Some of files we deal with aren't CSVs, so the parsing can be fairly complex and take longer than the disk access. However, the methodology is the same no matter what file type.
You can process pieces of the large files concurrently. Here's pseudo code of how we do it:
import os, multiprocessing as mp
# process file function
def processfile(filename, start=0, stop=0):
if start == 0 and stop == 0:
... process entire file...
else:
with open(file, 'r') as fh:
fh.seek(start)
lines = fh.readlines(stop - start)
... process these lines ...
return results
if __name__ == "__main__":
# get file size and set chuck size
filesize = os.path.getsize(filename)
split_size = 100*1024*1024
# determine if it needs to be split
if filesize > split_size:
# create pool, initialize chunk start location (cursor)
pool = mp.Pool(cpu_count)
cursor = 0
results = []
with open(file, 'r') as fh:
# for every chunk in the file...
for chunk in xrange(filesize // split_size):
# determine where the chunk ends, is it the last one?
if cursor + split_size > filesize:
end = filesize
else:
end = cursor + split_size
# seek to end of chunk and read next line to ensure you
# pass entire lines to the processfile function
fh.seek(end)
fh.readline()
# get current file location
end = fh.tell()
# add chunk to process pool, save reference to get results
proc = pool.apply_async(processfile, args=[filename, cursor, end])
results.append(proc)
# setup next chunk
cursor = end
# close and wait for pool to finish
pool.close()
pool.join()
# iterate through results
for proc in results:
processfile_result = proc.get()
else:
...process normally...
Like I said, that's only pseudo code. It should get anyone started who needs to do something similar. I don't have the code in front of me, just doing it from memory.
But we got more than a 2x speed up from this on the first run without fine tuning it. You can fine tune the number of processes in the pool and how large the chunks are to get an even higher speed up depending on your setup. If you have multiple files as we do, create a pool to read several files in parallel. Just be careful no to overload the box with too many processes.
Note: You need to put it inside an "if main" block to ensure infinite processes aren't created.
Try benchmarking reading your file and parsing each CSV row but doing nothing with it. You ruled out disk access, but you still need to see if the CSV parsing is what's slow or if your own code is what's slow.
If it's the CSV parsing that's slow, you might be stuck, because I don't think there's a way to jump into the middle of a CSV file without scanning up to that point.
If it's your own code, then you can have one thread reading the CSV file and dropping rows into a queue, and then have multiple threads processing rows from that queue. But don't bother with this solution if the CSV parsing itself is what's making it slow.
The simplest approach would probably be to do all 30 files at once with your existing code -- would still take all day, but you'd have all the files done at once. (ie, "9 babies in 9 months" is easy, "1 baby in 1 month" is hard)
If you really want to get a single file done faster, it will depend on how your counters actually update. If almost all the work is just in analysing value you can offload that using the multiprocessing module:
import time
import multiprocessing
def slowfunc(value):
time.sleep(0.01)
return value**2 + 0.3*value + 1
counter_a = counter_b = counter_c = 0
def add_to_counter(res):
global counter_a, counter_b, counter_c
counter_a += res
counter_b -= (res - 10)**2
counter_c += (int(res) % 2)
pool = multiprocessing.Pool(50)
results = []
for value in range(100000):
r = pool.apply_async(slowfunc, [value])
results.append(r)
# don't let the queue grow too long
if len(results) == 1000:
results[0].wait()
while results and results[0].ready():
r = results.pop(0)
add_to_counter(r.get())
for r in results:
r.wait()
add_to_counter(r.get())
print counter_a, counter_b, counter_c
That will allow 50 slowfuncs to run in parallel, so instead of taking 1000s (=100k*0.01s), it takes 20s (100k/50)*0.01s to complete. If you can restructure your function into "slowfunc" and "add_to_counter" like the above, you should be able to get a factor of 24 speedup.
Read one file at a time, use all CPUs to run search_function():
#!/usr/bin/env python
from multiprocessing import Array, Pool
def init(counters_): # called for each child process
global counters
counters = counters_
def search_function (value): # assume it is CPU-intensive task
some conditions checking:
update the counter_a or counter_b or counter_c
counter[0] += 1 # counter 'a'
counter[1] += 1 # counter 'b'
return value, result, error
if __name__ == '__main__':
counters = Array('i', [0]*3)
pool = Pool(initializer=init, initargs=[counters])
values = (line.split()[3] for line in textfile)
for value, result, error in pool.imap_unordered(search_function, values,
chunksize=1000):
if error is not None:
print('value: {value}, error: {error}'.format(**vars()))
pool.close()
pool.join()
print(list(counters))
Make sure (for example, by writing wrappers) that exceptions do not escape next(values), search_function().
Unfortunately, no. Reading in files and operating on the lines read (such as json parsing or computation) is a CPU-bound operation, so there's no clever asyncio tactics to speed it up. In theory one could utilize multiprocessing and multiple cores to read and process in parallel, but having multiple threads reading the same file is bound to cause major problems. Because your file is so large, storing it all in memory and then parallelizing the computation is also going to be difficult.
Your best bet would be to head this problem off at the pass by partitioning the data (if possible) into multiple files, which could then open up safer doors to parallelism with multiple cores. Sorry there isn't a better answer AFAIK.
There are several possibilites, but first profile your code in find the bottlenecks. Maybe your processing does some slows things which can be speed up - which would be vastly preferable to multiprocessing. If that does not help, you could try:
Use another file format. Reading serialized json from text is not the fastest operation in the world. So you could store your data (for example in hdf5) which could speed up processing.
Implement multiple worker processes which can read portions of the file (worker1 reads lines 0 - 1million, worker2 1million - 2million etc). You can orchestrate that with joblib or celery, depending on your needs. Integrating the results is the challenge, there you have to see what your needs are (map-reduce style?). This is more difficult in python due to no real threading than in other languages, so maybe you could switch the language for that.
Based on the comments here's example using multiprocessing.Pool.
Each process reads one file line by line and sends the result back to main process to collect.
import re
import multiprocessing
from collections import Counter
pat = re.compile(r"([A-Z]{2,})=(.+?)(?=[A-Z]{2,}=|$)")
def process_file(filename):
c = Counter()
with open(filename, "r") as f_in:
for line in f_in:
d = dict(pat.findall(line))
if "ADDRESS" in d:
c[d["ADDRESS"]] += 1
# send partial result back to main process:
return c
if __name__ == "__main__":
# you can get filelist for example from `glob` module:
files = ["file1.txt", "file2.txt"]
final_counter = Counter()
with multiprocessing.Pool(processes=4) as pool:
# iterate over files and update the final Counter:
for result in pool.imap_unordered(process_file, files):
final_counter.update(result)
# print final Counter:
for k, v in final_counter.items():
print("{:<20} {}".format(k, v))
Prints:
123 Fake St 2
234 Fake St 2
Note: You can use tqdm module to get nice progress bar, for example:
...
from tqdm import tqdm
...
for result in pool.imap_unordered(process_file, tqdm(files)):
...
Solution
Redis Queue RQ is likely the most appropriate datastore for this use-case. This will allow you to process the files in parallel while managing resources without needing to use multiprocessing—while you still could if you really needed to.
The benefit with this approach, in contrast with solely using multiprocessing or async is that it scales horizontally when used in conjunction with Redis Cluster.
Pseudocode
from redis import Redis
from rq import Queue, Retry
def do_work(f):
if maximum_filesize_in_queue_exceeds_resources():
raise ResourcesUnavailableError
else:
f.read()
...
def run(files_to_process):
q = Queue()
for f in files_to_process:
with open(filename, "r") as f:
q.enqueue(do_work, f, retry=Retry(max=3, interval=[10, 30, 60]))
if __name__ == "__main__":
files_to_process = ("foo", "bar", "baz")
run(files_to_process)
...
Results
You could use Redis to also store the results from each worker, and use rq's success hook to update the summary, which would also be stored in Redis. Once all queues have completed—you could then print a report with result and summary information stored in Redis.
References
https://redis.com
https://python-rq.org
https://redis.io/docs/manual/scaling/
It may be possible to parallelize this to speed it up, but doing multiple reads in parallel is unlikely to help.
Your OS is unlikely to usefully do multiple reads in parallel (the exception is with something like a striped raid array, in which case you still need to know the stride to make optimal use of it).
What you can do, is run the relatively expensive string/dictionary/list operations in parallel to the read.
So, one thread reads and pushes (large) chunks to a synchronized queue, one or more consumer threads pulls chunks from the queue, split them into lines, and populate the dictionary.
(If you go for multiple consumer threads, as Pappnese says, build one dictionary per thread and then join them).
Hints:
- ... push chunks to a synchronized queue ...
- ... one or more consumer threads ...
Re. bounty:
C obviously doesn't have the GIL to contend with, so multiple consumers are likely to scale better. The read behaviour doesn't change though. The down side is that C lacks built-in support for hash maps (assuming you still want a Python-style dictionary) and synchronized queues, so you have to either find suitable components or write your own. The basic strategy of multiple consumers each building their own dictionary and then merging them at the end is still likely the best.
Using strtok_r instead of str.split may be faster, but remember you'll need to manage the memory for all your strings manually too. Oh, and you need logic to manage line fragments too. Honestly C gives you so many options I think you'll just need to profile it and see.
It does seem tempting to think that using a processing pool will solve problems like this, but it's going to end up being a good bit more complicated than that, at least in pure Python.
Because the OP mentioned that the lists on each input line would be longer in practice than two elements, I made a slightly-more-realistic input file using :
paste <(seq 20000000) <(seq 2 20000001) <(seq 3 20000002) |
head -1000000 > largefile.txt
After profiling the original code, I found the slowest portion of the process to be the line-splitting routine. (.split() took approximately 2x longer than .append() on my machine.)
1000000 0.333 0.000 0.333 0.000 {method 'split' of 'str' objects}
1000000 0.154 0.000 0.154 0.000 {method 'append' of 'list' objects}
So I factored the split into another function and use a pool to distribute the work of splitting the fields :
import sys
import collections
import multiprocessing as mp
d = collections.defaultdict(list)
def split(l):
return l.split()
pool = mp.Pool(processes=4)
for keys in pool.map(split, open(sys.argv[1])):
d[keys[0]].append(keys[1:])
Unfortunately, adding the pool slowed things down by more than 2x. The original version looked like this :
$ time python process.py smallfile.txt
real 0m7.170s
user 0m6.884s
sys 0m0.260s
versus the parallel version :
$ time python process-mp.py smallfile.txt
real 0m16.655s
user 0m24.688s
sys 0m1.380s
Because the .map() call basically has to serialize (pickle) each input, send it to the remote process, and then deserialize (unpickle) the return value from the remote process, using a pool in this way is much slower. You do get some improvement by adding more cores to the pool, but I'd argue that this is fundamentally the wrong way to distribute this work.
To really speed this up across cores, my guess is that you'd need to read in large chunks of the input using some sort of fixed block size. Then you could send the entire block to a worker process and get serialized lists back (though it's still unknown how much the deserialization here will cost you). Reading the input in fixed-size blocks sounds like it might be tricky with the anticipated input, however, since my guess is that each line isn't necessarily the same length.
No one can really answer whether using ThreadPoolExecutor or ProcessPoolExecutor will be faster without knowing exactly what each task does. you need to try both and Benchmark the time taken by each to find which is better.
this code can help you figure that out yourself, it is based on this answer, but it uses a queue to limit the lines being read, so you don't risk having the entire file in memory if the processing is slow. also writing to the output file is done by its own thread, reading and writing to files (IO) releases the GIL, so they can both happen in parallel.
import concurrent.futures
import os
import queue
import threading
from io import IOBase
import time
from typing import Optional
def process_line(line: str):
# Simulate some CPU-bound work on the line
for i in range(int(1e6)):
pass
return line.upper()
def writer_task(out_file: IOBase, writer_queue: queue.Queue):
while True:
fut: Optional[concurrent.futures.Future] = writer_queue.get()
if fut is None:
break
line = fut.result()
out_file.write(line)
print("line written")
# Wrap main script behavior in main function
def main():
t1 = time.time()
with open("large_file.txt") as infile, open("output.txt", "w") as outfile:
with concurrent.futures.ThreadPoolExecutor() as executor:
writer_queue = queue.Queue(maxsize=os.cpu_count() * 2 + 10)
writer = threading.Thread(target=writer_task, args=(outfile, writer_queue), daemon=True)
writer.start()
for line in infile:
print("line read")
writer_queue.put(executor.submit(process_line, line))
writer_queue.put(None) # signal file end
writer.join()
t2 = time.time()
print(f"time taken = {t2-t1}")
# Invoke main function only when run as script, not when imported or invoked
# as part of spawn-based multiprocessing
if __name__ == '__main__':
main()
you can easily swap ThreadPoolExecutor for ProcessPoolExecutor and Measure which one is better. you might want to delete the print("line written") and its counterpart as they are only for illustrative purpose.
for something as small as just line.upper, then just processing it on the main thread will be faster than either option.
FYI: don't use this code in production, if an exception happens in the writer then your app will be stuck forever, you need to catch whatever fut.result() could throw.
If you're working with a multi-gigabyte file and have requirements for memory efficiency, order preservation, and concurrency, you can achieve this using Python's multiprocessing library. Use multiprocessing.Pool for parallel processing since this scenario is CPU-bound.