Functional programming and big data

Volume, velocity and variety, why big data needs distributed processing, and how immutability, statelessness and higher-order functions make MapReduce work.

A13.7Functional programmingA level25 min

A single BugBot logs a few hundred readings a minute. A fleet of ten thousand robots, each logging every sensor, all day, makes more data than one computer can store, let alone process. In module A11 you met big data as a database problem. This lesson looks at it as a programming problem, and at why the answer the industry settled on is functional programming.

What makes data big

Big data is a catch-all term for data that will not fit the usual containers: it is too big, arrives too fast or is too varied for one server or a relational database. It is usually described by three features:

• Volume: there is too much of it to store on, or process with, a single server.
• Velocity: it arrives so fast, often as a continuous stream, that it must be processed as it comes in, sometimes within milliseconds.
• Variety: it comes in many forms, such as text, images, sensor readings and video, often unstructured, so it does not fit into the rows and columns of a relational database.

Its size and lack of structure make it hard to analyse with ordinary queries, so machine learning techniques are used to find patterns in it.

Big data is often stored with a fact-based model: each fact records a single piece of information, such as "bot 7 read 31.0 cm at 14:02:07". Facts are never updated or deleted, only added, each with a timestamp, so the data is a growing record of everything that has been true. That is immutability again. The structure of such a dataset can be described by a graph schema: nodes for the things (a robot, a mat, a school), edges for the relationships between them (bot 7 drives on mat 3), and properties holding the values stored about a node (bot 7's battery level).

When one machine is not enough

If the data will not fit on one server, the processing must be distributed across many machines: the data is split into chunks, each machine works on its own chunk at the same time, and the results are combined. Writing that kind of program in the imperative style is very hard to get right:

• If two machines can change the same data, one can read a value halfway through the other's change. Preventing that needs locks, and locks bring deadlocks and slow everything down.
• If a function's result depends on state built up by earlier calls, then the order in which machines do the work changes the answer, and a chunk cannot simply be handed to whichever machine is free.
• If one of a thousand machines fails partway through, work that changed shared state cannot just be run again, because running it twice would change the state twice.

Why functional programming fits

Functional programming makes it much easier to write distributed code that is correct and efficient, because of three of its features:

This is also why the programs are easier to test: a pure function that works on your laptop on ten readings works identically on a cluster on ten billion.

MapReduce

The best-known pattern is MapReduce, which frameworks such as Hadoop and Spark are built around. It is map and fold, spread across a cluster:

1. Split: the data is divided into chunks, each stored on a different machine.
2. Map: every machine applies the same pure function to every record in its own chunk, at the same time as the others.
3. Reduce: the partial results are combined with a fold into one answer.

For the combining to give the same answer however the chunks were split and in whatever order the results arrive, the reduce function must be associative (and in practice commutative): combining (a with b) then c must equal a with (b with c). Adding counts is; subtracting is not.

from functools import reduce

CHUNKS = (
    (31.0, 51.0, 15.0),          # machine A's readings
    (44.0, 12.5),                # machine B's
    (60.2, 19.0, 33.2, 47.8),    # machine C's
)

def summarise(chunk):            # map stage: pure, runs on one machine
    return (len(chunk), reduce(lambda a, b: a + b, chunk, 0))

def combine(p, q):               # reduce stage: associative
    return (p[0] + q[0], p[1] + q[1])

partial = tuple(map(summarise, CHUNKS))
count, total = reduce(combine, partial)
print(partial)
print("mean of", count, "readings:", round(total / count, 2))

Run this in the simulator

Each machine sends back only a pair, (how many, total), not its readings, so very little data crosses the network. Because summarise is pure and combine is associative, the answer is the same as working through all nine readings on one machine. Note that the machines cannot send back their means and average those: the mean of means is wrong when the chunks are different sizes.

Task: count the fleet's events

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)

Challenges

1. Is "keep the larger of two numbers" associative? Is subtraction? For each, work out (8 with 5) with 2 and 8 with (5 with 2), and say which could be used as a reduce stage.
2. Change the program to count events per robot instead. Which function changes, and which stay exactly the same?
3. A fleet of robots streams readings at 10,000 per second. Which of volume, velocity and variety does that describe, and why can a relational database struggle with it?