There's quite a bit I don't understand in your code. So instead I'll just give you code that works ;-)
I'm baffled by how your code can run at all. A .gz file is compressed binary data (gzip compression). You should need to open it with Python's gzip.open(). As is, I expect it to die with an encoding exception, as it does when I try it.
temp[2] is not an integer. It's a string. You're not adding integers here, you're catenating strings with +. int() needs to be applied first.
I don't believe I've ever seen asyncio mixed with concurrent.futures before. There's no need for it. asyncio is aimed at fine-grained pseudo-concurrency in a single thread; concurrent.futures is aimed at coarse-grained genuine concurrency across processes. You want the latter here. The code is easier, simpler, and faster without asyncio.
While concurrent.futures is fine, I'm old enough that I invested a whole lot into learning the older multiprocessing first, and so I'm using that here.
These ngram files are big enough that I'm "chunking" the reads regardless of whether running the serial or parallel version.
collections.Counter is much better suited to your task than a plain dict.
While I'm on a faster machine than you, some of the changes alluded to above have a lot do with my faster times.
I do get a speedup using 3 worker processes, but, really, all 3 were hardly ever being utilized. There's very little computation being done per line of input, and I expect that it's more memory-bound than CPU-bound. All the processes are fighting for cache space too, and cache misses are expensive. An "ideal" candidate for coarse-grained parallelism does a whole lot of computation per byte that needs to be transferred between processes, and not need much inter-process communication at all. Neither are true of this problem.
from time import perf_counter as now
import gzip
from collections import Counter
PATH = r".\googlebooks-eng-all-1gram-20120701-a.gz"
def chunked_read(f, byte_limit=10**9):
while True:
lines = f.readlines(byte_limit)
if lines:
print("returning", format(len(lines), ','), "lines")
yield lines
else:
break
def crunch(lines):
c = Counter()
for line in lines:
temp = line.split('\t')
c[temp[0]] += int(temp[2])
return c
def show_counter(tag, c):
print(tag, f"{len(c) = :,} {c.total() = :,}")
if __name__ == "__main__":
if 1:
start = now()
word_count_dict = Counter()
with gzip.open(PATH, "rt", encoding='utf-8') as f:
for lines in chunked_read(f):
word_count_dict += crunch(lines)
print(f'total time used for sync version was {now() - start}')
show_counter("word_count_dict", word_count_dict)
del lines
if 1:
import multiprocessing as mp
MAXWORKERS = 3
start = now()
d = Counter()
with gzip.open(PATH, "rt", encoding='utf-8') as f, \
mp.Pool(MAXWORKERS) as p:
for r in p.imap_unordered(crunch,
chunked_read(f, 100_000_000)):
d += r
print(f'total time used for mp version was {now() - start}')
show_counter("d", d)
assert word_count_dict == d # verify same result either way
and output from one run:
returning 48,118,459 lines
returning 38,500,046 lines
total time used for sync version was 73.44404479999866
word_count_dict len(c) = 1,440,378 c.total() = 88,179,952,324
returning 4,813,692 lines
returning 4,816,691 lines
returning 4,807,902 lines
returning 4,814,039 lines
returning 4,806,524 lines
returning 4,812,883 lines
returning 4,808,835 lines
returning 4,812,116 lines
returning 4,811,212 lines
returning 4,814,568 lines
returning 4,811,115 lines
returning 4,811,647 lines
returning 4,818,276 lines
returning 4,813,439 lines
returning 4,819,090 lines
returning 4,813,714 lines
returning 4,815,252 lines
returning 4,797,510 lines
total time used for mp version was 59.9362641000007
d len(c) = 1,440,378 c.total() = 88,179,952,324
EDIT: Using concurrent.futures instead
It's really neither easier nor harder to use concurrent.futures here instead of multiprocessing, although mp has so very many bells & whistles it may take a while to realize that. The timing doesn't really matter either: under the covers, the time is overwhelmingly consumed by OS-level inter-process pipes or sockets. Which higher-level API you use to get at that doesn't much matter to speed. Just replace the mp part like so:
import concurrent.futures as cf
...
with gzip.open(PATH, "rt", encoding='utf-8') as f, \
cf.ProcessPoolExecutor(max_workers=MAXWORKERS) as ex:
for r in ex.map(crunch, chunked_read(f, 100_000_000)):
d += r
Indeed, the code is essentially the same.
Q&A
"MapReduce" is a term of art with several meanings, referring mostly to a way of thinking about arranging parallel computations, and especially to the Apache Hadoop implementation of that model. The standard (python.org) Python distribution does not support it directly as such.
In the code I showed you, the "map" part is spelled by functions with "map" in their name (multiprocessing's imap_unordered(), and concurrent.future's map()). The "reduce" part was spelled simply "+=" - Counters directly support that way to combine two of them, which is (or should be) more obvious and more efficient than any indirect way of doing it.
If you want to pursue MapReduce seriously, I suggest downloading elaborate software packages specifically aiming at that framework. At the extreme (like Hadoop), they're going to want to implement their own filesystem.
The "as complete" part from asyncio isn't really needed here either. Pool's imap_unordered() (which my code used) directly accomplishes that, returning results as they're completed. While I didn't show it here, concurrent.futures offers an as_completed() function that does much the same. Again since the packages implement those directly, they're more obvious and efficient than any indirect way of doing it.
I do aim to discourage you from using asyncio unless it's truly needed (and it isn't, here). It's a safe bet that a large majority of Python programmers have no idea what asyncio does, and so can't follow, debug, or extend code using it. multiprocessing and concurrent.futures are challenging enough on their own ;-)
About IPC (inter-process communication), your OS guarantees that two processes have different address spaces. Nothing is shared between them. That's as true of Python processes as of any others. If, e.g., you have a string object in a multi-process Python program, it's visible to only one process. If you want another process to see it, you need IPC to "send" the object to that other process. That's far from free. Under the covers, the string is turned into a sequence of bytes by the pickle module, that sequence of bytes is pushed into an OS-level socket or pipe inter-process connection, and the receiving process has to take that sequence and unpickle it to reconstruct a copy of the original string object.
In this problem, a huge bottleneck is the input data file. It's compressed binary data, and so there's no way to tell different processes offsets into the file so they can all start on their own piece. Instead one process has to decompress it and use IPC to send (again, physical copies of!) lines to other processes. After unzipping, that's closing on 2 gigabytes of raw data all on its own to send across processees, which they all chew over one byte at a time, through multiple layers of software (pickling, both ends of a pipe or socket, unpickling). There's a lot of machinery under the covers to support this, none of which works for free, including too ways of using helper threads and inter-process locks to ensure all the data stays sane and doesn't "freeze" the sending or receiving processes.
What's an "ideal" application? For example, you have a list of a million large integers you want to factor (into primes). A 200-digit integer is only about 200 bytes long. Sending that across processes isn't free, but the cost is trivial compared to the years it may take to factor it ;-) If instead you wanted to add 1 to each integer, the cost of sending an integer across processes is far higher than the cost of adding 1.
In the case at hand, the actual work done consists of breaking a short line into a handful of pieces (splitting on tabs), converting a string representing a small integer into an int, then adding that int to a running total in a dict indexed by another piece of the string. Not free, but not expensive either. It's not really a promising candidate for multiprocessing unless the input file is reworked so that multiple processes can read from it independently (for example, if the data was stored to begin with as 100 different files - then processes wouldn't need to communicate at all except to combine their dicts at the end).
