Seven Concurrency Models in Seven Weeks (2014)

---

Figure 21. Using a combiner

In our case, our reducer works just as well as a combiner, but some algorithms will require a separate combiner. Hadoop does not guarantee use of a combiner if one is provided, so we need to make sure that our algorithm doesn’t depend on whether, or how often, it is used.

Day 1 Wrap-Up

That’s it for day 1. In day 2 we’ll see how to use Hadoop to construct the batch layer of the Lambda Architecture.

Joe asks:

Is It All About Speed?

It’s tempting to think that all Hadoop gives us is speed—allowing us to process large bodies of data more quickly than we could on a single machine, and certainly that’s a very important benefit. But there’s more to it than that:

· When we start talking about clusters of hundreds of machines, failure stops being a risk and becomes a likelihood. Any system that failed when one machine in the cluster failed would rapidly become unworkable. For that reason, Hadoop has been constructed from the ground up to be able to handle and recover from failure.

· Related to the preceding, we need to consider not only how to retry tasks that were in progress on a failed node, but also how to avoid loss of data if one or more discs fails. By default, Hadoop uses the Hadoop distributed file system (HDFS), a fault-tolerant distributed file system that replicates data across multiple nodes.

· Once we start talking about gigabytes or more of data, it becomes unreasonable to expect that we’ll be able to fit intermediate data or results in memory. Hadoop stores key/value pairs within HDFS during processing, allowing us to create jobs that process much larger datasets than will fit in memory.

Taken together, these aspects are transformative. It’s no coincidence that this chapter is the only one in which we’ve executed our Wikipedia word-count example on an entire Wikipedia dump—MapReduce is the only technology we’re going to cover that realistically allows that quantity of data to be processed.

What We Learned in Day 1

Breaking a problem into a map over a data structure followed by a reduce operation makes it easy to parallelize. MapReduce, in the sense we’re using the term in this chapter, specifically means a system that efficiently and fault-tolerantly distributes jobs constructed from maps and reduces over multiple machines. Hadoop is a MapReduce system that does the following:

· It splits input between a number of mappers, each of which generates key/value pairs.

· These are then sent to reducers, which generate the final output (typically also a set of key/value pairs).

· Keys are partitioned between reducers such that all pairs with the same key are handled by the same reducer.

Day 1 Self-Study

Find

· The documentation for Hadoop’s streaming API, which allows MapReduce jobs to be created in languages like Ruby, Python, or Perl

· The documentation for Hadoop’s pipes API, which allows MapReduce jobs to be created in C++

· A large number of libraries build on top of the Hadoop Java API to make it easier to construct more complex MapReduce jobs. For example, you might want to take a look at Cascading, Cascalog, or Scalding.

Do

· While a word-count job is running, kill one of the machines in the cluster (not the master—Hadoop is unable to recover if the master fails). Examine the logs while you do so to see how Hadoop retries the work that was assigned to that machine. Verify that the results are the same as for a job that doesn’t experience a failure.

· Our word-count program does what it claims, but it’s not very helpful if we want to answer the question “What are the top 100 most commonly used words on Wikipedia?” Implement a secondary sort (the Internet has many articles about how to do so) to generate fully sorted output from our word-count job.

· The “top-ten pattern” is an alternative way to solve the “most common words on Wikipedia” problem. Create a version of our word-count program that uses this pattern.

· Not all problems can be solved by a single MapReduce job—often it’s necessary to chain multiple jobs, with the output of one forming the input for the next. One example is the PageRank algorithm. Create a Hadoop program that calculates the page rank of each Wikipedia page. How many iterations does it take before the results stabilize?

Day 2: The Batch Layer

Yesterday we saw how we could use Hadoop to parallelize across a cluster of machines. MapReduce can be used to solve a huge range of problems, but today we’re going to concentrate on how it fits into the Lambda Architecture.

Before we look at that, however, let’s consider the problem that the Lambda Architecture exists to solve—what’s wrong with traditional data systems?

Problems with Traditional Data Systems

Data systems are nothing new—we’ve been using databases to answer questions about the data stored within them for almost as long as computers have existed. Traditional databases work well up to a point, but the volume of data we’re trying to handle these days is pushing them beyond the point where they can cope.

Scaling

Some techniques enable a traditional database to scale beyond a single machine (replication, sharding, and so on), but these become harder and harder to apply as the number of machines and the query volume grows. Beyond a certain point, adding machines simply doesn’t help.

Maintenance Overhead

Maintaining a database spread over a number of machines is hard. Doing so without downtime is even more so—if you need to reshard your database, for example. Then there’s fault tolerance, backup, and ensuring data integrity, all of which become exponentially more difficult as the volume of data and queries increases.

Complexity

Replication and sharding typically require support at the application layer—your application needs to know which replicas to query and which shards to update (which will typically vary from query to query in nonobvious ways). Often many of the facilities that programmers have grown used to, such as transaction support, disappear when a database is sharded, meaning that programmers have to handle failures and retries explicitly. All of this increases the chances of mistakes being made.

Human Error

An often-forgotten aspect of fault tolerance is coping with human error. Most data corruptions don’t result from a disk going bad, but rather from a mistake on the part of either an administrator or a developer. If you’re lucky, this will be something that you spot quickly and can recover from by restoring from a backup, but not all errors are this obvious. What if you have an error that results in widespread corruption that goes undetected for a couple of weeks? How are you going to repair your database?

Sometimes you can undo the damage by understanding the effects of the bug and then creating a one-off script to fix up the database. Sometimes you can undo it by replaying from log files (assuming your log files capture all the information you need). And sometimes you’re simply out of luck. Relying on luck is not a good long-term strategy.

Reporting and Analysis

Traditional databases excel at operational support—the day-to-day running of the business. They’re much less effective when it comes to reporting and analysis, both of which require access to historical information.

A typical solution is to have a separate data warehouse that maintains historical data in an alternative structure. Data moves from the operational database to the data warehouse through a process known as extract, transform, load (ETL). Not only is this complicated, but it depends upon accurately predicting which information you’ll need ahead of time—it’s not at all uncommon to find that some report or analysis you would like to perform is impossible because the information you would need to run it has been lost or captured in the wrong structure.

In the next section we’ll see how the Lambda Architecture addresses all these issues. Not only does it allow us to handle the vast quantities of data modern applications are faced with, but it also does so simply, recovering from both technical failure and human error and maintaining the complete historical record that will enable us to perform any reporting or analysis we might dream up in the future.

Eternal Truths

We can divide information into two categories—raw data and derived information.

Consider a page on Wikipedia—pages are constantly being updated and improved, so if I view a particular page today, I may well see something different from what I saw yesterday. But pages aren’t the raw data from which Wikipedia is constructed—a single page is the result of combining many edits by many different contributors. These edits are the raw data from which pages are derived.

Furthermore, although pages change from day to day, edits don’t. Once a contributor has made an edit, that edit never changes. Some subsequent edit might modify or undo its effect, and therefore the derived page, but edits themselves are immutable.

You can make the same distinction in any data system. The balance of your bank account is derived from a sequence of raw debits and credits. Facebook’s friend graph is derived from a sequence of raw friend and unfriend events. And like Wikipedia edits, both debits and credits and friend and unfriend events are immutable.

This insight, that raw data is eternally true, is the fundamental basis of the Lambda Architecture. In the next section we’ll see how it leverages that insight to address the problems of traditional data systems.

Joe asks:

Is All Raw Data Really Immutable?

At first it can be difficult to see how some kinds of raw data could be eternally true. What about a user’s home address, for example? What happens if that person moves to a different house?

This is still immutable—we just need to add a timestamp. Instead of recording “Charlotte lives at 22 Acacia Avenue,” we record that “On March 1, 1982, Charlotte lived at 22 Acacia Avenue.” That will remain true, whatever happens in the future.

Data Is Better Raw

At this point, your ears should be pricking up. As we’ve seen in previous chapters, immutability and parallelism are a marriage made in heaven.

An Appealing Fantasy

Let’s allow ourselves to fantasize briefly. Imagine that you had an infinitely fast computer that could process terabytes of data in an instant. You would only ever hold on to raw data—there would be no point keeping track of any of the information derived from it, because we could derive it as and when we needed it.

At a stroke, in this fantasy land we’ve eliminated most of the complexity associated with a traditional database, because when data is immutable, storing it becomes trivial. All our storage medium needs to do is allow us to append new data as and when it becomes available—we don’t need elaborate locking mechanisms or transactions, because once it’s been stored it will never change.

It gets better. When data is immutable, multiple threads can access it in parallel without any concern of interfering with each other. We can take copies of it and operate on those copies, without worrying about them becoming out-of-date, so distributing the data across a cluster immediately becomes much easier.

Of course, we don’t live in this fantasy land, but you might be surprised how close we can get by leveraging the power of MapReduce.

Joe asks:

What About Deleting Data?

Occasionally we have good reasons to delete raw data. This might be because it’s outlived its usefulness, or it might be for regulatory or security reasons (data-protection laws may forbid retention of some data beyond a certain period, for example).

This doesn’t invalidate anything we’ve said so far. Data we choose to delete is still eternally true, even if we choose to forget it.

Fantasy (Almost) Becomes Reality

If we know ahead of time which queries we want to run against our raw data, we can precompute a batch view, which either directly contains the derived data that will be returned by those queries or contains data that can easily be combined to create it. Computing these batch views is the job of the Lambda Architecture’s batch layer.

As an example of the first type of batch view, consider building Wikipedia pages from a sequence of edits—the batch view will simply comprise the text of each page, built by combining all the edits of that page.

The second type of batch view is slightly more complex, so that’s what we’ll concentrate on for the remainder of today. We’re going to use Hadoop to build batch views that will allow us to query how many edits a Wikipedia contributor has made over a period of days.

Wikipedia Contributors

The kind of query that we’d ideally like to make is, “How many contributions did Fred Bloggs make between 3:15 p.m. on Tuesday, June 5, 2012, and 10:45 a.m. on Thursday, June 7, 2012?” To do so, however, we’d need to maintain and index a record of the exact time of every contribution. If we really need to make this kind of query, then we’ll need to pay the price, but in reality we’re unlikely to need to make queries at this fine a granularity—a day-by-day basis is likely to be more than enough.

So our batch view could consist of simple daily totals:

This would work well enough if we were always interested in periods of a few days, but queries for periods of several months would still require combining many values (potentially, for example, 365 to determine how many contributions a user had made in a year). We can decrease the amount of work required to answer this kind of query by keeping track of periods of both months and days:

This would allow us to decrease the amount of work required to count a user’s contributions within a year from summing 365 values to 12. And we can handle periods that neither start nor finish at the beginning of a month by summing monthly values and either adding or subtracting daily values:

Contributor Logging

Sadly, we don’t have access to a live feed of Wikipedia contributors. But if we did, it might look something like this:

	2012-09-01T14:18:13Z 123456789 1234 Fred Bloggs
	2012-09-01T14:18:15Z 123456790 54321 John Doe
	2012-09-01T14:18:16Z 123456791 6789 Paul Butcher
	⋮

The first column is a timestamp, the second is an identifier representing the contribution, the third is an identifier representing the user who made the contribution, and the remainder of the line is the username.

Although Wikipedia doesn’t publish such a feed, it does provide periodic XML dumps containing a full history (you’re looking for enwiki-latest-stub-meta-history).^[69] The sample code for this chapter includes an ExtractWikiContributors project that will take one of these dumps and create a file of the preceding form.

In the next section we’ll construct a Hadoop job that takes these log files and generates the data required for our batch view.

Counting Contributions

As always, our Hadoop job consists of a mapper and a reducer. The mapper is very straightforward, simply parsing a line of the contributor log and generating a key/value pair in which the key is the contributor ID and the value is the timestamp of the contribution:

LambdaArchitecture/WikiContributorsBatch/src/main/java/com/paulbutcher/WikipediaContributors.java
	public static class Map extends Mapper<Object, Text, IntWritable, LongWritable> {
	public void map(Object key, Text value, Context context)
	throws IOException, InterruptedException {
	Contribution contribution = new Contribution(value.toString());
	context.write(new IntWritable(contribution.contributorId),
	new LongWritable(contribution.timestamp));
	}
	}

Most of the work is done by the Contribution class:

LambdaArchitecture/WikiContributorsBatch/src/main/java/com/paulbutcher/Contribution.java
Line 1	class Contribution {
-	static final Pattern pattern = Pattern.compile("^([^\\s]*) (\\d*) (\\d*) (.*)$");
-	static final DateTimeFormatter isoFormat = ISODateTimeFormat.dateTimeNoMillis();
-
5	public long timestamp;
-	public int id;
-	public int contributorId;
-	public String username;
-
10	public Contribution(String line) {
-	Matcher matcher = pattern.matcher(line);
-	if(matcher.find()) {
-	timestamp = isoFormat.parseDateTime(matcher.group(1)).getMillis();
-	id = Integer.parseInt(matcher.group(2));
15	contributorId = Integer.parseInt(matcher.group(3));
-	username = matcher.group(4);
-	}
-	}
-	}

We could parse a log file line in various ways—in this case, we’re using a regular expression (line 2). If it matches, we use the ISODateTimeFormat class from the Joda-Time library to parse the timestamp and convert it to a long value representing the number of milliseconds since January 1, 1970 (line 13).^[70] The contribution and contributor IDs are then just simple integers, and the contributor’s username is the remainder of the line.

Our reducer is more involved:

LambdaArchitecture/WikiContributorsBatch/src/main/java/com/paulbutcher/WikipediaContributors.java
Line 1	public static class Reduce
-	extends Reducer<IntWritable, LongWritable, IntWritable, Text> {
-	static DateTimeFormatter dayFormat = ISODateTimeFormat.yearMonthDay();
-	static DateTimeFormatter monthFormat = ISODateTimeFormat.yearMonth();
5
-	public void reduce(IntWritable key, Iterable<LongWritable> values,
-	Context context) throws IOException, InterruptedException {
-	HashMap<DateTime, Integer> days = new HashMap<DateTime, Integer>();
-	HashMap<DateTime, Integer> months = new HashMap<DateTime, Integer>();
10	for (LongWritable value: values) {
-	DateTime timestamp = new DateTime(value.get());
-	DateTime day = timestamp.withTimeAtStartOfDay();
-	DateTime month = day.withDayOfMonth(1);
-	incrementCount(days, day);
15	incrementCount(months, month);
-	}
-	for (Entry<DateTime, Integer> entry: days.entrySet())
-	context.write(key, formatEntry(entry, dayFormat));
-	for (Entry<DateTime, Integer> entry: months.entrySet())
20	context.write(key, formatEntry(entry, monthFormat));
-	}
-	}

For each contributor, we build two HashMaps, days (line 8) and months (line 9). We populate these by iterating over timestamps (remember that values will be a list of timestamps) using the Joda-Time utility methods withTimeAtStartOfDay and withDayOfMonth to convert that timestamp to midnight on the day of the contribution and midnight on the first day of the month (lines 12 and 13, respectively). We then increment the relevant count in days and months using a simple utility method:

LambdaArchitecture/WikiContributorsBatch/src/main/java/com/paulbutcher/WikipediaContributors.java
	private void incrementCount(HashMap<DateTime, Integer> counts, DateTime key) {
	Integer currentCount = counts.get(key);
	if (currentCount == null)
	counts.put(key, 1);
	else
	counts.put(key, currentCount + 1);
	}

Finally, once we’ve finished building our maps, we iterate over each map and generate an output for each day and month in which there was at least one contribution (lines 17 to 20).

This is slightly involved because, as we’ve seen, the output from a Hadoop job is always a set of key/value pairs, but what we want to output are three values—the contributor ID, a date (either a month or a day), and a count. We could do this by defining a composite value. This way, our key is the contributor ID and the value is a composite value containing the date and the count. But our case is simple enough that we can instead just use a string as the value type and format it appropriately with formatEntry:

LambdaArchitecture/WikiContributorsBatch/src/main/java/com/paulbutcher/WikipediaContributors.java
	private Text formatEntry(Entry<DateTime, Integer> entry,
	DateTimeFormatter formatter) {
	return new Text(formatter.print(entry.getKey()) + "\t" + entry.getValue())
	}

Here’s a section of this job’s output:

	463 2001-11-24 1
	463 2002-02-14 1
	463 2001-11-26 6
	463 2001-10-01 1
	463 2002-02 1
	463 2001-10 1
	463 2001-11 7

This contains exactly the data that we need, but a collection of text files isn’t particularly convenient. In the next section we’ll talk about the serving layer, which indexes and combines the output of the batch layer.

Joe asks:

Can We Generate Batch Views Incrementally?

The batch layer we’ve described so far recomputes entire batch views from scratch each time it’s run. This will certainly work, but it’s probably performing more work than necessary—why not update batch views incrementally with the new data that’s arrived since the last time the batch view was generated?

The simple answer is that there’s nothing to stop you from doing so, and this can be a useful optimization. But you can’t rely exclusively on incremental updates—much of the power of the Lambda Architecture derives from the fact that we can always rebuild from scratch if we need to. So feel free to implement an incremental algorithm if the optimization is worth the additional effort, but recognize that this can never be a replacement for recomputation.

Completing the Picture

The batch layer isn’t enough on its own to create a complete end-to-end application. That requires the next element of the Lambda Architecture—the serving layer.

The Serving Layer

The batch view we’ve just generated needs to be indexed so that we can make queries against it, and we need somewhere to put the application logic that decides how to combine elements of the batch view to satisfy a particular query. This is the duty of the serving layer:

We’re going to leave the serving layer as an exercise for the reader, as it has little relevance to the subject matter of this book, but it’s worth mentioning one particular aspect of it—the database.

Although you could build the serving layer on top of a traditional database, its access patterns are rather different from a traditional application. In particular, there’s no requirement for random writes—the only time the database is updated is when the batch views are updated, which requires a batch update.

Therefore a category emerges of serving-layer databases optimized for this usage pattern, the most well-known of which are ElephantDB and Voldemort.^[71][72]

Almost Nirvana

Taken together, the batch and serving layers give us a data system that addresses all the problems we identified at the beginning of the day:

The batch layer runs in an infinite loop, regenerating batch views from our raw data. Each time a batch run completes, the serving layer updates its database.

Because it only ever operates on immutable raw data, the batch layer can easily exploit parallelism. Raw data can be distributed across a cluster of machines, enabling batch views to be recomputed in an acceptable period of time even when dealing with terabytes of input.

The immutability of raw data also means that the system is intrinsically hardened against both technical failure and human error. Not only is it much easier to back up raw data, but if there’s a bug, the worst that can happen is that batch views are temporarily incorrect—we can always correct them by fixing the bug and recomputing them.

Finally, because we retain all raw data, we can always generate any report or analysis that might occur to us in the future.

There’s an obvious problem, though—latency. If the batch layer takes an hour to run, then our batch views will always be at least an hour out-of-date. This is where tomorrow’s subject, the speed layer, comes into play.

Day 2 Wrap-Up

This brings us to the end of day 2. In day 3 we’ll complete our picture of the Lambda Architecture by looking at the speed layer.

What We Learned in Day 2

Information can be divided into raw data and derived information. Raw data is eternally true and therefore immutable. The batch layer of the Lambda Architecture leverages this to allow us to create systems that are

· highly parallel, enabling terabytes of data to be processed;

· simple, making them both easier to create and less error prone;

· tolerant of both technical failure and human error; and

· capable of supporting both day-to-day operations and historical reporting and analysis.

The primary drawback of the batch layer is latency, which the Lambda Architecture addresses by running the speed layer alongside.

Day 2 Self-Study

Find

· The approaches we’ve discussed here are not the only way to tackle building a data system that leverages Hadoop—other options include HBase, Pig, and Hive. All three of these have more in common with a traditional data system than what we’ve seen today. Pick one and compare it to the Lambda Architecture’s batch layer. When might you choose one, and when the other?

Do

· Finish the system we built today by creating a serving layer that takes the output of the batch layer, puts it in a database, and allows queries about the number of edits made by a particular user over a range of days. You can build it either on top of a traditional database or on top of ElephantDB.

· Extend the preceding to build batch views incrementally—to do this, you will need to provide the Hadoop cluster with access to the serving layer’s database. How much more efficient is it? Is it worth the additional effort? For which types of application would incremental batch view construction make sense? For which would it not?

Day 3: The Speed Layer

As we saw yesterday, the batch layer of the Lambda Architecture solves all the problems we identified with traditional data systems, but it does so at the expense of latency. The speed layer exists to solve that problem. The following figure shows how the batch and speed layers work together: