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.

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.

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.

To fully utilize your hardware core, it's better to use the multiprocessing library.

from multiprocessing import Pool

from os import listdir 
import csv

def process_file(file):
    #load the text file as list using csv module 
    #run a bunch of operations
    #regex the int from the filename. for ex file1.txt returns 1, and file42.txt returns 42
    #write out a corresponsding csv file in dirB. For example input file file99.txt is written as out99.csv

if __name__ == '__main__':
    mypath = "some/path/"

    inputDir = mypath + 'dirA/'
    outputDir = mypath + 'dirB/'

    p = Pool(12)
    p.map(process_file, listdir(inputDir))

Document of multiprocessing: https://docs.python.org/2/library/multiprocessing.html

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)):

...

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.

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.

The python standard library offers two process pools to simplify what you're trying to achieve: multiprocessing.Pool (as mentioned by @RootTwo) and concurrent.futures.ProcessPoolExecutor.

I personally like the second one (despite its more limited API) because you can replace it with a ThreadPoolExecutor without changing the rest of the code. Threads are better suited if your processing is I/O bound, because inter-process communication is slow and processes are slower to start (especially on Windows). With the concurrent.futures API, you can easily swap the two and test which one works better.

There are some other ways you can improve the code, most notably:

1. If you're on Python 3.x as you should be, I suggest the use of pathlib instead of os.path.
2. Put code from the if __name__ == "__main__": into a "main" function to make automated testing easier.
3. Use snake_case instead of camelCase as that's the accepted standard for Python.
4. (Opinion) If you're on python 3.x, you could also use type annotations to improve the readability of the code.

With this in mind, I would rewrite your code as:

from pathlib import Path
from concurrent.futures import ProcessPoolExecutor, as_completed
from typing import Iterable

def get_files(directory: Path) -> Iterable[Path]:
    # Using glob simplifies the code in these cases.
    return (file for file in directory.glob("**/*") if file.is_file())

def foo(file: Path):
    ...

# This can be easily called from a test
def process_files(directory: Path):
    # I am using sum, so that the result is computed lazily and we
    # do not need to build a list of all files. If the directory is
    # very large, this could save a lot of memory.
    # Since get_files returns a one-shot generator, we cannot
    # save it to a variable and reuse it here and below.
    file_count = sum(1 for _ in get_files(directory))
    
    with concurrent.futures.ProcessPoolExecutor() as executor:
        futures = executor.map(foo, get_files(directory))
        # Reading the value returned by the executor is very
        # important, because exceptions happening in the `foo`
        # function will be captured and not be raised until the
        # value is read, thus obfuscating the errors.
        #        
        # Another nice solution for the progress would be
        # using the library tqdm.
        for i, _ in enumerate(as_completed(futures)):
            print(f"Processed file {i+1} / {file_count}")

if __name__ == "__main__":
    process_files(Path(""))