Imagine giving a stack of text files to several workers: each worker reads its assigned chunk and turns words into intermediate key-value pairs. That first, independent stage is the map phase. In a word-count computation, it reads files, splits them into words, and emits pairs such as ("the", 1) for a later reduce phase. section 13.10
A map phase typically involves:
- Partitioning input into chunks so multiple workers can process files concurrently; with M workers, the design aims for M chunks. section 13.10
- Applying a map function that parses each assigned file and writes intermediate key-value results. section 13.14
- Optionally combining local results before the reduce phase to aggregate output and reduce communication overhead. section 13.14
- Accepting nondeterministic file assignment, which allows more files to be read simultaneously as worker nodes are added. section 13.10
Word counting
For input containing “the take the,” the map phase can emit [("the", 1), ("take", 1), ("the", 1)]. The reduce phase later combines values belonging to the same key, producing [("the", 2), ("take", 1)]. section 13.10
# Chapter 13/wordcount/server.py
import os
import glob
import asyncio
from scheduler import Scheduler
from protocol import Protocol, HOST, PORT, FileWithId
class Server(Protocol):
def __init__(self, scheduler: Scheduler) -> None:
super().__init__()
self.scheduler = scheduler
def connection_made(self, transport: asyncio.Transport) -> None: #1
peername = transport.get_extra_info("peername") #1
print(f"New worker connection from {peername}") #1
self.transport = transport #1
self.start_new_task() #1
def start_new_task(self) -> None: #2
command, data = self.scheduler.get_next_task() #2
self.send_command(command=command, data=data) #2
def process_command(self, command: bytes,
data: FileWithId = None) -> None:
if command == b"mapdone":
self.scheduler.map_done(data)
self.start_new_task()
elif command == b"reducedone":
self.scheduler.reduce_done()
self.start_new_task()
else:
print(f"Unknown command received: {command}")
def main():
event_loop = asyncio.get_event_loop() #3
current_path = os.path.abspath(os.getcwd())
file_locations = list( #4
glob.glob(f"{current_path}/input_files/*.txt")) #4
scheduler = Scheduler(file_locations) #5
server = event_loop.create_server(
lambda: Server(scheduler), HOST, PORT) #6
server = event_loop.run_until_complete(server) #7
print(f"Serving on {server.sockets[0].getsockname()}")
try: #8
event_loop.run_forever() #8
finally: #8
server.close() #8
event_loop.run_until_complete(server.wait_closed()) #8
event_loop.close() #8
if __name__ == "__main__":
main()
Inspect the server-side Python implementation that assigns map work and coordinates workers. section 13.14
