Google GFS and MapReduce

01 · CONTEXT

Two papers, one bet on commodity machines

GFS and MapReduce were born in the same environment: large clusters of commodity machines, where failure is routine. The GFS paper starts precisely from there, treating component failures as the norm, not the exception. The MapReduce paper describes clusters of hundreds or thousands of machines, in which machine failures are common.

The two also share a view of the workload. In GFS, files are huge by traditional standards and mostly change by appending new data, after which they are only read, often sequentially. MapReduce hides parallelization, fault tolerance, data distribution, and load balancing inside a library.

The link between the two is written in the MapReduce paper: the input data is managed by GFS and stored on the local disks of the cluster machines. This case shows how each paper solves one layer and how the decisions of one rest on those of the other. According to the same paper, the cluster's storage uses a distributed file system developed in-house.

02 · GFS

Metadata on the master, data on the chunkservers

A GFS cluster has a single master and several chunkservers, and is accessed by several clients. Files are divided into fixed-size chunks; each chunk is stored as a regular file on a chunkserver's local disk and is replicated across several of them, with three replicas by default.

The master keeps all the metadata, including the mapping from files to chunks and the current location of the chunks. The data itself never passes through it: the client asks the master where the chunk is and then talks directly to the chunkservers.

1. The client converts the file name and offset into a chunk index and sends the master the file name and that index.

2. The master responds with the chunk identifier and the location of the replicas.

3. The client caches the response and sends the request to one of the replicas, probably the closest.

4. The chunk's bytes flow between the chunkserver and the client; the file data does not pass through the master.

5. Subsequent reads of the same chunk do not go back to the master until the cached information expires or the file is reopened.

The large chunk is a deliberate choice. Reads and writes on the same chunk require only one initial request to the master for the location, and the large chunk reduces the size of the metadata, which makes it possible to keep it in memory. The paper states that the master keeps less than 64 bytes of metadata for each 64 MB chunk.

03 · THE SINGLE MASTER

What the single master buys and what it costs

The authors justify the single master by simplicity: it simplifies the design and enables sophisticated chunk placement decisions, and these decisions, of placement and replication, use global knowledge. The trade-off is that the master's involvement in reads and writes must be minimal so it does not become a bottleneck. The paper states it chose the centralized approach to simplify the design, increase reliability, and gain flexibility.

The master persists the namespace and the file-to-chunk mapping through an operation log on the local disk, replicated on remote machines, and this log is the only persistent record of the metadata. It does not store the chunk locations persistently: it asks the chunkservers at startup, and the paper explains that the chunkserver has the final word on which chunks exist on its disks.

For availability, the master's operation log and checkpoints are replicated on several machines. If the master's machine or disk fails, monitoring infrastructure external to GFS starts a new master process elsewhere using the replicated log. In addition, shadow masters provide read-only access to the file system even with the primary master down.

1. Suppose that one chunkserver, B, becomes unavailable.

2. The master tracks the chunkservers through periodic HeartBeat messages; when the number of available replicas falls below the target, it re-replicates the chunk.

3. The master instructs a chunkserver to copy the data directly from an existing valid replica; the copy flows between chunkservers.

In re-replication, the master instructs a chunkserver to copy the data directly from an existing valid replica.

04 · MAPREDUCE

Two user functions and a task master

The user writes two functions. Map takes an input pair and produces a set of intermediate key/value pairs; the library groups all intermediate values with the same key and passes them to the Reduce function, which combines them into a possibly smaller set. In the word count example, map emits each word with the occurrence count and reduce sums the counts emitted for each word.

The library splits the input into pieces and starts several copies of the program on the cluster. One of them is special, the master; the others are workers that receive work from it. The master picks idle workers and assigns each one a map or reduce task.

1. The user program splits the input and starts several copies of itself on the cluster; one of them is the master.

2. The master picks idle workers and assigns each one a map or reduce task.

3. Each map worker reads the contents of its input split, which is managed by GFS.

4. The intermediate pairs are periodically written to the map worker's local disk, partitioned into regions; the master receives the locations and forwards them to the reduce workers.

5. The reduce worker reads this data by remote call and sorts it by key.

6. The reduce worker appends the output of the Reduce function to an output file for its partition, in the global file system.

For each map and reduce task, the master stores the state and the identity of the worker machine, and it is the channel through which the location of the intermediate data reaches the reduce workers.

05 · HOW THEY COMPLEMENT EACH OTHER

Locality, failures, and output: where one paper fits into the other

Network bandwidth is a relatively scarce resource in the environment of the MapReduce paper. GFS stores several copies of each block on different machines, and the MapReduce master takes the location of the input files into account to schedule the map task on a machine that holds a replica. If it cannot, it tries to schedule near a replica, for example on a worker on the same network switch.

1. GFS stores copies of each block on different machines; here, machine A has a replica of the input block.

2. The MapReduce master considers the location of the input file and tries to schedule the map task on a machine that has a replica.

3. If it cannot, it tries to schedule the task near a replica, for example on a worker on the same network switch as the machine with the data.

4. When this holds for a good share of the cluster's workers, most of the input is read locally and consumes no network bandwidth.

Fault tolerance is also split by layer. In GFS, each chunk is replicated on several chunkservers, and the master re-replicates a chunk as soon as the number of available replicas falls below the user-defined target. In MapReduce, the functional model makes it possible to use re-execution as the main fault-tolerance mechanism: completed map tasks are re-executed after a failure because their output is on the local disk of the machine that failed, while completed reduce tasks do not need to be re-executed because their output is in a global file system.

There is also a point of contact on writes. MapReduce relies on the atomic rename operation of the underlying file system to guarantee that the final state contains only the data from one execution of the reduce task. On the GFS side, namespace mutations, such as file creation, are atomic and handled exclusively by the master, and a typical write pattern is to generate the file from start to finish and atomically rename it to the permanent name.

The MapReduce paper cites a use that brings the two layers together: in the production indexing system, the crawled documents are stored as files in GFS, and the process runs as a sequence of five to ten MapReduce operations.

06 · LIMITS AND LESSONS

What both pay for the single coordinator

In MapReduce, the master must make O(M + R) scheduling decisions. The paper cites a typical run with M equal to 200,000 and R equal to 5,000, using 2,000 worker machines, which shows the size of the coordination work.

Both systems have a single coordinator, and the papers handle its failure in different ways, as the previous table shows: GFS replicates the master's state and MapReduce, in the described implementation, aborts the computation.