MapReduce is a batch processing framework, which means that it starts with a fixed collection of data, runs a sequence of steps on it, then outputs an answer. If the data changes or new data is added after the process is started, you have to run the whole thing again from scratch in order to incorporate the changes. That’s why, for example, it used to take a couple of days for new websites to show up in Google search results; once a Google spider found a new page, it wouldn’t be added to the search results until the next time the PageRank algorithm was run on the entire set of known web pages. Nowadays, Google uses a streaming framework that allows search results to be updated incrementally. I’ll discuss a few different approaches to streaming data analysis in future posts.

Last time, I described the PageRank algorithm in terms of tubes that carried water between buckets representing different websites. One can think of the algorithm as keeping track of how much water would be poured into each bucket during each minute that the system runs. (The water pumped out of the bucket is proportional to the water level in the bucket, which is much easier to keep track of.) Because the number of websites that we would want to keep track of is too large for a single computer to handle, we will divide it up on a distributed network. Each computer will be responsible for keeping track of the water level and the outgoing links for a subset of the entire list of websites.

But a website recorded on one computer will often have a link/tube to a website stored on a different computer. In order for the second computer to correctly calculate the amount of water coming in to this website, the first computer will have to send it the information about this link/tube.

So, you can think of MapReduce as a system for organizing the messages between the different computers on the network about how the pieces of data that they’re responsible for interact with each other. There are a number of other types of problems that require a similar sort of messaging. The most common example used to explain MapReduce is an algorithm that counts how many times each word appears in a large collection of texts. The idea is to have each computer in the network be responsible for some number of text files, and some collection of words. Each computer makes a list of how many times each word appears in the texts that it’s responsible for. Then it has to send these totals to the computers responsible for each word, which calculates the final total.

The word count example is reminiscent of the basic distributed learning framework that I discussed two posts back, in the sense that each computer calculates partial sums that are sent to a single computer to calculate the total. The difference here is that instead of having one master computer that computes one final sum, we have multiple final sums, calculated by multiple computers.

The MapReduce framework formalizes this in terms of three steps: Map, Combine and Reduce. The Combine step is optional, and is in many cases the same as the Reduce step, which is why it gets left out of the name. We’ll think about the steps in terms of sending messages between the computers in the network. The Map step can be thought of as writing down the list of messages that need to get sent. The Combine step consolidates the messages so that there is less to send across the network (since this is usually the bottleneck.) The Reduce step then processes all the messages received by each computer.

The step that’s missing from this is, of course, the sorting and delivering. This is called the Shuffle step and is all handled by the Map/Reduce framework. In fact, you can think of MapReduce as a program that has been almost entirely written, except that three routines have been left out, namely Map, Combine and Reduce. All you have to do is to fill in those three functions so that they fit into the existing program. Lets see how we might do this for PageRank.

First, for the Map step, we want to create the messages that need to get sent out. In other words, for each website/bucket, we want to send a message to each website/bucket that it has a link/tube to and tell it how much much water is coming in along that tube. Many of these different websites will be stored on the same computer, possibly even the computer that’s currently doing the processing. But the Map routine doesn’t have to worry about that, since it’s handled by the shuffle step. We’ll address each message to the website/bucket, and let the MapReduce framework decide which computer on the network to send it to.

In practice, the input to each function might be a filename or a link to an entry in a database, either of which will have to be processed in some way to extract the data. But to make the following pseudo-code easier to read, I’ve assumed that the input is given directly.

// input: water_level - the current water level the site 
                           being processed
          links - a list of site ids that are linked to from the
                    current site
// output: a list, each member of which is 
             a pair (site id, message)

function Map(water_level, links)
    // the output is split evenly between the links
    output = (0.7 * water_level)/size(links)

    message_list = new List

    for each (link_id in links)
        message_list.add(link_id, output)

    return message_list

The MapReduce framework will call this function for each website in the data set on each computer, then send the resulting messages to the computers that are responsible for each link id. These computers, in turn, run the reduce function, which in the case of PageRank calculates the new water level for each site. Again, in practice the output of this function would have to be saved to disk or entered into a database, but to make the pseudocode simpler, I’m just making the new water level the return value of the Reduce function.

// input: water_level: the current water level of this site
          incoming: a list of incoming messages
// output: the new water level

function Reduce(water_level, incoming)
    new_level = water_level

    for each (water_amount in incoming)
        new_level += water_amount

    return new_level

Now, you may have noticed that I left out the Combine step. Recall that the combine step is meant to reduce the number of messages that need to be transmitted across the network. In particular, it’s worth noting that during the Map step, MapReduce doesn’t send the messages across the network as they’re produced. With networks, it’s almost always faster to send one long message than to send lots of short ones. So once all the Map steps have been completed on a given computer, the framework sorts them all based on what computer they’ll be sent to (usually using a trick such as a hash function) and then sends one big packet of messages to each computer. During this process, it also has the option of running the Combine step.

In our case, what would the Combine step do? Lets say that after running all the Map steps for the different websites that are stored on computer A, we end up with three different links to the same website on computer B. Well, rather than including these three different messages in the packet of messages sent to computer B, and letting computer B add them all up, it would make more sense to have computer A add them up, and just include one message in the packet. So, the Combine step will be essentially identical to the Reduce step – adding up all the values that are sent to each site.

This Combine step is possible because addition is associative; we get the same sum no matter what order we add the numbers in. Not every Reduce step will have this property, so not every MapReduce algorithm will have a Combine step (and theoretically some may have a Combine step that is significantly different from the Reduce step.) Thus the MapReduce framework will only run a Combine step if it’s explicitly told to do so.

Note that if we were to implement PageRank under MapReduce in this way, it would only give us one iteration. To get to the final answer, we would have to run through the Map-Combine-Shuffle-Reduce cycle multiple times, as well as adding in a way to keep track of when to stop.

In fact, many algorithms end up requiring different combinations and sequences of Map-Combine-Shuffle-Reduce steps, so it has become common to think of these as separate building blocks that can be arranged and stacked in lots of different ways. This is the idea behind Google’s FlumeJava, as described in this white paper (PDF). The FlumeJava library is not publicly available, but there is at least one open source clone of it.

But perhaps more importantly, echoes of these Map-Combine-Shuffle-Reduce steps appear in many of the newer distributed learning frameworks, even those that replace batch processing with streaming, and those that are specifically designed for graphs. I plan to describe some of these new frameworks in upcoming posts, and I’ll try to point out the hints of MapReduce that arise.

When you type a query into a search engine, the results that it returns generally come from a list that was created in advance from programs called “spiders” that download all the webpages they can find and record information about them. The process of finding all the sites that would match your search is fairly straightforward (at least at small scales – there are lots of technical issues that come up when you try to deal with with billions of pages.)

The tricky part turns out to be that there are usually so many links that would fit the search, that the search engine has to decide which ones to leave out. Or put another way, the problem is to decide (in an automated way) which links should go on the first page of search results and which should go later in the list. (Note that nowadays PageRank is only a small part of a much larger system that Google uses to sort its search results. Moreover, there are a huge number of “SEO experts” employed by other companies whose main job is to figure out this system and redesign web pages so they’ll appear higher on the list.)

Sorting search results was one of the major problems that plagued search engines in the early years of the web and this is where the PageRank algorithm came in. Because the computer can’t interpret the contents of each page the way a person can, PageRank relies on other indicators, namely how many other web pages have links to the web page in question. The assumption is that a web page that many other pages link to is probably important, and should be placed higher on the list. We can think of this in terms of giving each page a score, then sorting the pages so that the pages with the highest scores come first.

The simplest way to do this would be to simply count the number of incoming links to each page and let this number be its score – The page that fits the search terms and has the most incoming links goes first and so on. This would certainly give higher scores to pages with more incoming links, but it would also make it very easy for page designers to game the system. For example, someone might set up 1000 different nearly-identical web sites, all of which link to each other and have pages filled with advertisements, with a list of buzz words at the bottom. Since each of these pages would have 1000 incoming links, they would get a very high score, even though they don’t deserve it.

PageRank attempts to skirt this issue by taking into account not only how many links point to a given page, but how high the scores of the pages that link to it are. This seems like circular logic, though, since you need to know the scores of all the pages before you can calculate their scores. But it turns out you can deal with this issue using something like a feedback loop. If you don’t know what a feedback loop is, don’t worry. The idea is pretty simple.

Imagine that each web page is bucket of water with a pump underneath. Each link from webpage A to webpage B is a tube coming out of the pump underneath bucket A and pouring into bucket B. Three of these buckets are shown in the Figure below. If we fill all the buckets partway up with water, then turn on the pumps, the buckets with more tubes (i.e. links) leading into them will start to fill up, while the buckets with fewer links will start to run out of water. Note that water never enters or leaves the system. In fact, water that leaves a particular bucket will often come back to it at a later time. Or I should say that it will feed back into it. (Get it? Feed back.)

To make this work right, we’ll install special water pumps that speed up when the water level in the bucket rises and that slow down when the water level drops. If we do this correctly, then each bucket will, at some point, reach an equilibrium state where the amount of water removed by the pump is exactly equal to the amount of water coming in from the tubes. (The amount of water coming in from each tube will also reach an equilibrium as the water levels of the other buckets reach equilibrium.)

In other words, at some point the water levels in the different buckets will stop changing. At this point, the more water flowing into any given bucket, the higher its level will be (since a higher water level means the pump is pushing more water out, which is equal to the amount coming in.) So, when the system reaches this point, the web sites/buckets with lots of tubes coming into them will generally have high water levels, while the websites with fewer tubes coming in will have very low levels. We can then read off the each web site’s score from the water level in its bucket.

Note that the group of 1000 sites that all reference each other in the hopes of improving their ranking won’t fare too well under this system: Since they’re all sharing the same amount of water, their water levels will all stay the same. On the other hand, for a web site that doesn’t have a whole lot of links coming in, but these links mostly come from other websites that, in turn, have a lot of incoming links, this system will work relatively well, since the small number of tubes feeding into the bucket will still bring in a lot of water.

Now, as you might expect, the PageRank algorithm is not actually implemented using physical buckets and pumps. But we can think of the algorithm as simulating what would happen in such a system. This is related to a number of ideas in mathematics, such as differential equations and dynamical systems.

There are plenty of very nicely written accounts of how this is technically carried out, but as usual I want to look at it from a more geometric perspective. If you’ve been following this blog, you may be able to predict how we’re going to turn this into a geometry problem (and no, it has nothing to do with shape of the buckets, or the curvature of the tubes.)

Recall that the goal of the PageRank algorithm is to keep track of the water level in each bucket. So this gives us a list of numbers. Well, guess what that list of numbers is… It’s a vector(!) which we can think of as defining a point in some (high-dimensional) space.

In order to visualize the situation, lets consider what would happen if we just had three websites/buckets, so their three scores would give us a point in three-dimensional space. As we noted above, the amount of water in the system is always the same, so while the water levels in the three buckets will change, adding up all the water levels will always give us the same amount. In other words, at any point in time, if we add together the three coordinates of our vector, we’ll get the same number. In most discussions of PageRank, this sum is set to 1.

So, lets start by looking at all the hypothetically possible sets of water levels/scores/vectors. In other words, we want to consider the set consisting of each vector with positive (or, rather, non-negative) coordinates that add up to 1. As it turns out, this set forms a triangle with corners at the vectors (1,0,0), (0,1,0), (0,0,1), like the one shown in the Figure below. This shape is sometimes called the convex hull of these three points – the smallest convex set that contains them. If we did this with four websites, the shape would be a three-dimensional tetrahedron (the convex hull of four points) in four-dimensional space. For more websites, we would get higher-dimensional convex hulls in higher dimensional spaces.

Getting back to the three-bucket case, if we put water in, the water levels of the three buckets will define some point in this triangle. Imagine that we’ve put the tip of a pen at that point. If we turn on the pumps, the water levels will go up and down and the point will start to move, as its coordinates change. Because the total amount of water doesn’t change, the point will stay in the triangle. Moreover, if we move the tip of the pen along with the point, the pen will trace out some curve on the triangle.

There are two things to note about this thought experiment: First, while we originally said we would start with all the buckets having equal amounts of water (which defines the point at the very center of the triangle), that’s not completely necessary. We could start with any water levels, corresponding to any point in the triangle, then turn on the pumps, and the pen would trace out some curve. Three such (hypothetical) curves are shown in the Figure.

Second, (and this is a lot more subtle) if we stopped the pumps, moved the pen to a different location, then started them again, the second curve traced out by the pen would never cross the curve that was drawn the first time. The reason is that the direction that the pen moves only depends on the current water levels, not where it was a earlier. So if the pen ever gets to a point where it was previously, it will move in the same direction it moved the last time it was there. So if I had drawn more curves, they would look like neatly combed hair – they might be wavy, but they run parallel to each other.

In particular, at each point in the triangle, we can draw a small arrow that shows what direction the curve would go if we started the pen at that point. You can think of this arrow as representing a vector, in which case we call the set of all vectors that arise this way a vector field. (In fact, these arrows are tangent vectors of the curves, which have to do with derivatives, which is why you can think about this in terms of differential equations, but lets not worry about that now.)

If we were to start with all these arrows drawn on the triangle, we could figure out the curve that our system of buckets and pumps would draw by just following the arrows. As it turns out, if we do this right, there will be some point on the triangle (or on the higher dimensional shape if we do this with more buckets) where all the arrows nearby point in towards that point. This is shown on the right of the figure. Any curve that the pen draws would head towards this point. In fact, as the pen gets closer to this point, it will get slower and slower and never quite reach that point. (If Zeno were alive today he would love this.) This point where the paths all come together is the equilibrium where the water levels stop changing, and coordinates of this point define the scores that we want to assign to each of the webpages.

Exactly calculating one of these curves in order to find the equilibrium point, it turns out, is an unnecessarily complex problem. Instead, PageRank calculates an approximation of a curve broken up into a number of steps. It starts at the center of the shape defined by all possible water levels (the center of the triangle in our three-dimensional example) and calculates the direction of the arrow at that point. Then, instead of moving continuously in that direction, it jumps along a straight line along the arrow. Since the actual path might curve a little, this straight line will take the new point off the curve, but it will still be pretty close. Then the algorithm calculates the arrow at this new point, takes another jump and then repeats.

The path traced out by this sequence of jumps won’t follow any of the actual curves exactly, but that’s OK because of the way the arrows all (eventually) point towards the equilibrium point. So the sequence of points calculated by PageRank will also move in this direction. When it gets really close, it will start spiraling around the equilibrium point, taking smaller and smaller steps each time. When the steps get small enough, it means the point is near the equilibrium point and the algorithm stops, taking these final coordinates as a good approximation of the actual equilibrium.

Incidentally, this is very similar in philosophy to many of the gradient methods that I’ve discussed previously on this blog, including logistic regression and mixture models. All of these work by successively updating a set of numbers, and can be formulated in terms of following a path by jumping along a vector field (though I didn’t explain it that way in the earlier posts.)

The equations that determine how the point jumps towards the equilibrium point are pretty simple, and as I mentioned above, there are plenty of places you can read about the details. Where it gets complicated is the fact this has to be done with billions of websites and billions of links. This is far too much for a single computer to handle, so from the very first days of the PageRank algorithm, it has been carried out on distributed networks of computers. This required developing systems to deal with the fact a website stored on one machine in the network will usually link to other websites stored on lots of other machines across the network. In my next post, I’ll discuss one of the systems that was developed to handle this sort of problem, namely Map/Reduce.

Before we get into the basics of parallel/distributed machine learning, there’s one thing that I should point out: If you start with a poorly designed/poorly optimized algorithm, you will generally get a much bigger gain in efficiency by optimizing the algorithm than by just adding machines to your parallel cluster. So, before trying to parallelize an algorithm it’s best to first make sure that the conventional algorithm is designed as efficiently as possible.

The most common setup for parallel computing these days is to have a large number of individual machines, each with its own processor, connected to its own memory and hard drive(s). Each processor may have two or more “cores” that can work in parallel. However, unlike with traditional super computers, the processors in different machines do not have direct access to each other’s data. (Originally, the term super computer referred to machines with a large number of parallel processors and centralized, shared memory, though nowadays it’s also often applied to large clusters with distributed memory.) Because of the distributed memory, individual processors must explicitly communicate with each other, usually through ethernet or one of its faster relatives. This setup has the advantage that it is easy to add machines to a cluster, plus the machines that are used are relatively inexpensive (so called “commodity” machines.)

On the other hand, having so many (relatively cheap) machines means that you are essentially guaranteed to have some number of them not working properly at any given time. So, any software that runs on such a cluster has to include measures that allow it to keep running even if some of the machines fail. Usually, this means keeping multiple copies of data, regularly checking on the machines involved, and having procedures for switching parts of the process to healthy processors when necessary. But for this post, I’m going to gloss over this aspect.

Because each machine in a cluster has its own hard drive, it is common to store a (large) data set across multiple machines. Usually, each machine has a subset of the data points/rows of the data set. Because communication across ethernet is slow compared to communication between each processor and its local memory and hard drive, the goal is usually to have each processor access only its own data as much as possible. There will often be one machine that is designated as the leader, while the rest are workers. The general idea behind creating a parallel algorithm is to have each of the worker machines summarize its local data in some way (depending on the algorithm), and then send the summary to the lead machine in as concise a way as possible so that the lead can effectively combine all the different summaries that it receives.

You can think of the lead machine like the CEO of a big corporation who gets reports from all the managers below her. If the manager sends the CEO too much detail about the operations of their unit, the CEO won’t have the time to go through it all. If the manager sends too simple a message, the CEO won’t have enough information to effectively compare it to the reports from the other units. So the job of each manager who reports to the CEO is to find the right balance between detail and brevity.

A distributed algorithm has to find the exact same balance. Depending on the algorithm, there will be some minimal amount of information that the lead processor needs to get from each worker. However, any information beyond this will make the algorithm slower.

To get an idea of how this can work in practice, I want to look at principal components analysis (PCA). We’ll think of the algorithm slightly differently than how I described it in the previous post, by breaking it into four steps. First, we find the centroid/average of the data. Second, we translate the data by subtracting this centroid from each data point. Third, we calculate the covariance matrix. Fourth, we calculate the eigenvectors of the covariance matrix. We’ll look at how to implement each of these steps in turn.

For the first step, calculating the centroid/average, we need to add up the vectors defined by all the data points and then divide by the number of data points. The nice thing about this is that addition (vector addition as well as scalar addition) has the associative property. This means that if we add the numbers together in groups, then add up the sums of the groups, we’ll get the same answer as if we just added them one by one. So to find the average on a distributed cluster, we have each worker add up all its data points, then report to the master the total that it got and the number of data points that it added together. The master processor then adds each of these intermediate sums together to get the complete total. It then adds up the total number of data points, and divides to get the centroid.

For step two, the master sends the centroid vector to each worker machine. Each worker is then responsible for translating its own data points.

Step three gets a little tricky because calculating the covariance matrix involves multiplying two other matrices. In the conventional algorithm, we have to calculate where is the matrix whose rows are the data points and whose columns are the features (after subtracting the centroid from each row). The matrix is the transpose matrix that we get from switching the rows and columns of . In general, multiplying two matrices requires us to multiply together numbers from different parts of the two matrices. This could be problematic if each machine in our cluster only has access to certain parts of the matrix.

But if we look closely at this particular calculation, , it turns out that we can do it in a way that only requires each machine to use its locally stored data. We carry out the matrix multiplication by multiplying numbers from by numbers from , then adding them up in a specified way. But we always multiply numbers from column i of by numbers in row i of . Because is the transpose of , this means that we’re always multilpying together two numbers from the same data vector.

In particular, we can redefine the calculation of  as follows: For each data vector , we will calculate the matrix in which the number in row i, column j is the product of the number in dimension i of and the number in dimension j. If we calculate this matrix for each data point (i.e. each row in ) and add them all up, the result will be exactly .

So we can carry out step three in much the same way as step one: Each processor calculates for each of its stored locally data points, then adds them all together to form a single matrix. Each processor then sends this one matrix to the master, which adds them all together. Because matrix addition, like vector addition, is associative, the final sum is exactly .

For the final step, we have to calculate the eigenvectors of this matrix. However, note that the size of this final matrix depends on the number of features, rather than the number of data points. The usual assumption when dealing with distributed data like this is that the number of features will remain a manageable size, while the number of data points will grow too large for a single processor. Since the size of the final matrix is determined by the number of features, it will usually be small enough that we can just use a conventional algorithm on the lead machine for the last step.

So, in the end, we repeatedly used one trick, in slightly different forms, to make PCA parallel. Chu et. al. [1] have shown that you can use this same trick – having workers compute partial sums and then sending them to the master for the total sum – for a wide range of algorithms, including many of the algorithms that I’ve discussed previously on this blog.

Interestingly, one of the harder algorithms to adapt to a distributed cluster is the back-propagation method for training a neural network. The problem is that, by design, this algorithm requires adapting the weights of the neural network to one data point at a time. You can’t process the next data point until you’ve finished processing the current one, so you can’t have multiple processors work simultaneously. One way to fix this issue is a technique called batch training, in which you use multiple data points each time you update the weights. There are also some more advanced techniques for pre-training a neural network that make the back-propagation step go faster, and many of these techniques are easier to adapt to a distributed system. But that’s a topic for a future post.

[1] Map-reduce for machine learning on multicore, CT Chu, SK Kim, YA Lin, YY Yu, G Bradski, AY Ng, K Olukotun, NIPS, 2006. (The link has a lapsed security certificate but seems to be safe.)