A13.7 Functional programming and big data

Three machines each hold one chunk of a fleet's event log (in the starter). Each line is <robot>,<event>, such as "bot2,bump". Count how many times each event happens, MapReduce style: - to_counts(line) returns a dictionary with one key, the event, and the value 1: to_counts("bot2,bump") is {"bump": 1}. - merge(a, b) takes two dictionaries of counts and returns a new dictionary holding every event in either, with the counts added. It must not change a or b, so there must be no d[key] = ... assignment or update anywhere in the program: build the new dictionary with a dictionary comprehension. - machine(chunk) is what one machine does: map to_counts over the chunk's lines and fold the results together with merge, starting from {}. - Map machine over CHUNKS, then fold the three partial results together with merge. Print the totals one per line, events in alphabetical order, as <event> <count>, such as bump 4. Then print same as one machine: True if the result equals machine applied to all the lines joined into one tuple (it should).

# the two lines every program starts with: the commands, then the robot
from bugbot import *
connect()
from functools import reduce

CHUNKS = (
    ("bot1,bump", "bot2,tag", "bot1,tag", "bot3,stall"),
    ("bot2,bump", "bot3,tag", "bot1,bump"),
    ("bot3,tag", "bot2,stall", "bot1,tag", "bot2,bump"),
)

counts = {}
for chunk in CHUNKS:
    for line in chunk:
        event = line.split(",")[1]
        counts[event] = counts.get(event, 0) + 1
print(counts)

The hint students can ask for: Test merge on its own first with two small dictionaries, including an event that is only in one of them. The keys of the new dictionary are every key in either. Once merge works, machine is one fold, and so is combining the machines.

A solution

# the two lines every program starts with: the commands, then the robot
from bugbot import *
connect()
from functools import reduce

CHUNKS = (
    ("bot1,bump", "bot2,tag", "bot1,tag", "bot3,stall"),
    ("bot2,bump", "bot3,tag", "bot1,bump"),
    ("bot3,tag", "bot2,stall", "bot1,tag", "bot2,bump"),
)

def to_counts(line):
    return {line.split(",")[1]: 1}

def merge(a, b):
    return {k: a.get(k, 0) + b.get(k, 0) for k in set(a) | set(b)}

def machine(chunk):
    return reduce(merge, map(to_counts, chunk), {})

totals = reduce(merge, map(machine, CHUNKS), {})
print("\n".join(map(lambda k: k + " " + str(totals[k]), sorted(totals))))

all_lines = reduce(lambda a, b: a + b, CHUNKS, ())
print("same as one machine:", machine(all_lines) == totals)

Any program that meets the task's checks is marked correct in the simulator; this is one way, not the only way.