Big data

Volume, velocity and variety; why one server is not enough; distributed processing with map and reduce; the fact-based model and graph schema.

A11.9Databases and big dataA level60 min

One robot's run log fits in a small table. Now imagine every BugBot in every school sending its sensor readings to one place, ten times a second, with camera images and voice clips alongside. No single database server could store that, let alone query it while it keeps arriving. Big data is the name for data like this, and it needs different ways of storing and processing data from the relational databases of this module.

Volume, velocity and variety

Big data is data that cannot be stored, processed or analysed on a single server using conventional database methods. It is usually described by three properties:

• Volume: there is too much data to fit on one server, or to process on one in a useful time.
• Velocity: the data arrives fast and continuously, as a stream, and often has to be processed as it arrives, within seconds or milliseconds.
• Variety: the data comes in many forms: structured (tables), semi-structured (JSON, XML) and unstructured (images, audio, video, free text). Unstructured data does not fit the rows and columns of a relational table, and this lack of structure is often the hardest part.

robots = 1000
readings_per_second = 10
bytes_per_reading = 64

per_day = robots * readings_per_second * bytes_per_reading * 60 * 60 * 24
print(f"readings a day: {robots * readings_per_second * 86400:,}")
print(f"storage a day: {per_day / 10**9:.1f} GB")
print(f"storage a year: {per_day * 365 / 10**12:.1f} TB")

Run this in the simulator

A thousand robots sending a small reading ten times a second produce 864 million readings and about 55 GB a day, before a single camera image. That is volume and velocity together; the images and audio add variety.

Finding anything useful in data this large and this messy cannot be done by writing a query for each question. Machine learning techniques are used to find the patterns: which sensor readings come before a robot gets stuck, for example.

Why one server is not enough

A relational database is designed for one server: a fixed schema of normalised tables, joins, and ACID transactions. For big data each of those becomes a problem. Unstructured data has no schema to fit. A single server can only be made so big (scaling up), so the data has to be spread across many machines (scaling out). Joins and transactions that span many machines are slow and hard to keep correct.

Distributed processing

In distributed processing the data is split across many computers, called nodes, and each node processes its own part at the same time. Since the data is too big to move, the program is sent to the data. Each part is also copied to more than one node, so the job survives when a node fails, which, among thousands of nodes, happens every day.

A common pattern is MapReduce:

1. Map: on every node, a map function turns each record into zero or more (key, value) pairs. It looks at one record only.
2. Shuffle: the system gathers all the pairs with the same key together, from every node.
3. Reduce: a reduce function combines the values for each key into a result.

from functools import reduce

# the run log, split across three nodes: (robot, cm)
nodes = [
    [("Ada", 42.0), ("Bolt", 38.0)],
    [("Ada", 80.0), ("Cog", 76.5)],
    [("Bolt", 81.0), ("Ada", 45.5)],
]

def mapper(record):
    robot, cm = record
    return (robot, cm)

mapped = [list(map(mapper, node)) for node in nodes]      # each node maps its own records
print("mapped:", mapped)

shuffled = {}
for pairs in mapped:
    for key, value in pairs:
        shuffled.setdefault(key, []).append(value)
print("shuffled:", shuffled)

totals = {key: reduce(lambda a, b: a + b, values) for key, values in shuffled.items()}
print("reduced:", totals)

Run this in the simulator

Programming this way suits functional programming (module A13). A map function with no side effects, working on data it never changes, gives the same answer whichever node runs it and in whatever order, so the work can be split and repeated safely. Functions such as map, filter and reduce, which take other functions as arguments, are exactly the shape of the job.

The fact-based model

A relational database overwrites: UPDATE robots SET colour = 'blue' destroys the fact that Ada was green. Big data systems often use a fact-based model instead:

• data is stored as facts, each recording a single piece of information;
• facts are immutable: they are never updated or deleted, only added;
• each fact is timestamped, so two otherwise identical facts at different times are different facts;
• the current state is not stored but worked out from the facts when needed.

facts = [
    ("09:00", "Ada", "colour", "green"),
    ("09:05", "Ada", "team", "Hawks"),
    ("10:30", "Ada", "colour", "blue"),
]

def value_at(robot, attribute, time):
    latest = None
    for ts, who, what, value in sorted(facts):
        if who == robot and what == attribute and ts <= time:
            latest = value
    return latest

print("Ada's colour at 10:00:", value_at("Ada", "colour", "10:00"))
print("Ada's colour now:", value_at("Ada", "colour", "23:59"))
print("facts kept:", len(facts))

Run this in the simulator

The model has real advantages. The complete history is kept, so the system can answer questions about any moment in the past. It tolerates human error: a program that writes wrong data adds wrong facts but destroys nothing, and the mistake can be corrected by adding facts or ignoring the bad ones. Adding to the end of a data set is also the simplest thing to spread across many nodes. The cost is storage, and the work of deriving the current state, which systems reduce by precomputing common views.

Graph schema

A graph schema describes how the facts connect, as a graph:

• nodes are the entities, such as a robot or a team;
• edges are the relationships between them, drawn as labelled arrows;
• properties are the data attached to a node, drawn here in rectangles joined by dashed lines.

A graph suits data where the connections matter most, such as "which students booked a robot that ran the square", because a query simply follows the edges rather than joining tables. It also stretches easily: a new kind of node or edge can be added without redesigning every table.

Task: map, shuffle, reduce

Three servers each hold part of the robots' event log. servers is a list of three lists; every item is a string "<robot>,<event>", where event is bump or stall.

1. Write mapper(line): it takes one line and returns a list of (key, value) pairs, [(robot, 1)] if the event is bump, or [] otherwise.
2. Call mapper on every line of every server, and shuffle: group the values by key.
3. Write reducer(key, values): it takes a robot name and the list of its values, and returns (key, total).
4. Call reducer once for each robot, in name order, and print each result as <robot> <bumps>, such as Cog 2.

The robot does not move.

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

# each server holds part of the event log: "robot,event"
servers = [
    ["Ada,bump", "Bolt,bump", "Ada,stall"],
    ["Cog,bump", "Ada,bump", "Bolt,stall", "Bolt,bump"],
    ["Ada,bump", "Cog,bump"],
]

Task: what was true then

facts is a list of immutable facts (timestamp, robot, attribute, value): a whole number, then three strings. They arrived out of order.

1. Write state_at(facts, t): it takes the facts and a whole number t and returns the state at time t (you choose the structure): for each robot and attribute, the value of the fact with the latest timestamp no later than t. It must not change the list: no assigning to its items, and no sort, append, insert, remove, pop or del on it.
2. For t equal to 4, then 10, print every robot and attribute in the state, sorted by robot then attribute, as at <t>: <robot> <attribute> <value>, such as at 4: Bolt colour red.

Seven lines in all. The robot does not move.

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

# (timestamp, robot, attribute, value): facts arrive in any order and are never changed
facts = [
    (5, "Ada", "colour", "blue"),
    (2, "Bolt", "colour", "red"),
    (9, "Ada", "team", "Owls"),
    (1, "Ada", "colour", "green"),
    (6, "Bolt", "team", "Owls"),
    (3, "Ada", "team", "Hawks"),
    (8, "Bolt", "colour", "yellow"),
]

Challenges

1. Change the MapReduce task to find each robot's total of all events, and then the most common event overall. Which function changed?
2. A wrong fact says Bolt joined Hawks at time 7. Without deleting anything, how could the system correct it?
3. Draw a graph schema for a social network of students who follow each other and like robot videos.