{"id":9,"date":"2011-06-17T15:31:19","date_gmt":"2011-06-17T19:31:19","guid":{"rendered":"https:\/\/michaelnielsen.org\/ddi\/?p=9"},"modified":"2011-06-18T10:14:34","modified_gmt":"2011-06-18T14:14:34","slug":"pregel","status":"publish","type":"post","link":"https:\/\/michaelnielsen.org\/ddi\/pregel\/","title":{"rendered":"Pregel"},"content":{"rendered":"

In this post, I describe a simple but powerful framework for distributed computing called Pregel<\/em>. Pregel was developed by Google, and is described in a 2010 paper<\/a> written by seven Googlers. In 2009, the Google Research blog<\/a> announced that the Pregel system was being used in dozens of applications within Google. <\/p>\n

Pregel is a framework oriented toward graph-based algorithms<\/em>. I won’t formally define graph-based algorithms here – we’ll see an example soon enough – but roughly speaking a graph-based algorithm is one which can be easily expressed in terms of the vertices of a graph, and their adjacent edges and vertices. Examples of problems which can be solved by graph-based algorithms include determining whether two vertices in a graph are connected, where there are clusters of connected vertices in a graph, and many other well-known graph problems. As a concrete example, in this post I describe how Pregel can be used to determine the PageRank<\/a> of a web page.<\/p>\n

What makes Pregel special is that it’s designed to scale very easily on a large-scale computer cluster. Typically, writing programs for clusters requires the programmer to get their hands dirty worrying about details of the cluster architecture, communication between machines in the cluster, considerations of fault-tolerance, and so on. The great thing about Pregel is that Pregel programs can be scaled (within limits) automatically<\/em> on a cluster, without requiring the programmer to worry about the details of distributing the computation. Instead, they can concentrate on the algorithm they want to implement. In this, Pregel is similar to the MapReduce<\/a> framework. Like MapReduce, Pregel gains this ability by concentrating on a narrow slice of problems. What makes Pregel interesting and different to MapReduce is that it is well-adapted to a somewhat different class of problems.<\/p>\n

Using Pregel to compute PageRank<\/h3>\n

A Pregel program takes as input a graph, with many vertices and (directed) edges. The graph might, for example, be the link graph of the web, with the vertices representing web pages, and the edges representing links between those pages. Each vertex is also initialized with a value. For our PageRank example, the value will just be an initial guess for the PageRank.<\/p>\n

Since I’m using PageRank as an example, let me briefly remind you how PageRank works. Imagine a websurfer surfing the web. A simple model of surfing behaviour might involve them either: (1) following a random link from the page they’re currently on; or (2) deciding they’re bored, and “teleporting” to a completely random page elsewhere on the web. Furthermore, the websurfer chooses to do the first type of action with probability 0.85, and the second type of action with probability 0.15. If they repeat this random browsing behaviour enough times, it turns out that the probability they’re on a given webpage eventually converges to a steady value, regardless of where they started. This probability is the PageRank for the page, and, roughly speaking, it’s a measure of the importance of the page: the higher the PageRank, the more important the page.<\/p>\n

I won’t get into any more of the ins-and-outs of PageRank, but will assume that you’re happy enough with the description above. If you’re looking for more details, I’ve written an extended introduction<\/a>.<\/p>\n

We’re going to use Pregel to compute PageRank. During the initialization stage for the graph we’ll start each vertex (i.e., webpage) off with an estimate for its PageRank. It won’t make any difference to the final result what that starting estimate it, so let’s just use a very easy-to-compute probability distribution, say, initializing each vertex with a value that is just one over the total number of web pages (i.e., the number of vertices).<\/p>\n

Once the input graph is initialized, the Pregel computation proceeds through a series of supersteps<\/em>. During each superstep each vertex does two things: (1) it updates its own value; and (2) it can send messages to adjacent vertices. The way it updates its own value is by computing a user-specified function which depends on the value of the vertex at the end of the previous superstep, as well as the messages sent to the vertex during the last superstep. Similarly, the messages the vertex sends can depend both on its value and the messages it was sent during the last superstep.<\/p>\n

To make this more concrete, here’s some python-like pseudocode showing how Pregel can be used to compute the PageRank of a link graph: <\/p>\n

\r\nimport pregel # includes pregel.Vertex class which we'll subclass\r\n\r\nclass PageRankVertex(pregel.Vertex):\r\n\r\n    def update(self):\r\n        if self.superstep < 50:\r\n            # compute a new estimate of PageRank, based on the most recent\r\n            # estimated PageRank of adjacent vertices\r\n            self.value = 0.15 * 1\/num_vertices + 0.85 * sum(self.messages)\r\n            # send the new estimated PageRank to adjacent vertices, dividing\r\n            # it equally between vertices\r\n            self.messages = [(self.value \/ num_adjacent_vertices) \r\n                for each adjacent vertex]\r\n        else:\r\n            # stop after 50 supersteps\r\n            self.active = False\r\n\r\ninitialize link structure for a 10-vertex graph\r\nnum_vertices = 10\r\nfor each vertex: # initialize values in the graph\r\n    vertex.value = 1\/num_vertices\r\nrun pregel\r\noutput values for all vertices\r\n<\/pre>\n
 The code should be self-explanatory: we initialize the graph, with each vertex starting with an initial estimate for its own PageRank. We update that by imagining the websurfer making a single step, either teleporting to a new random vertex, or else randomly following a link to one of the adjacent vertices.  This is repeated many times - I've arbitrarily chosen 50 supersteps - before halting.<\/p>\n
With the PageRank pseudocode under our belts, let's return to the bigger picture and summarize the basic Pregel model a bit more formally.<\/p>\n
Input phase:<\/strong> The input to a Pregel computation is a set of vertices.  Each vertex has an initial value, and may also have an associated set of outgoing edges.<\/p>\n
Computation phase:<\/strong> This is split up into supersteps.  In any given superstep each vertex can update its value and also its outgoing edges.  It can also emit messages, which are sent to adjacent vertices.  The updated value, outgoing edges and messages are determined by a user-defined function of the vertice's value, edges, and incoming messages at the start of the superstep.<\/p>\n
Note that Google's Pregel system allows messages to be sent to any<\/em> other vertex, not just adjacent vertices, although the paper implies that in most cases messages are usually sent only to adjacent vertices.  Also, Google's Pregel system allows both vertices and edges to have values.  I've omitted both these for simplicity in the code below, although both are easily restored.<\/p>\n
Halting:<\/strong> Each vertex has an attribute which determines whether it is active<\/tt> or not.  Vertices start active<\/tt>, but can change to inactive<\/tt> at any superstep.  The computation halts when every vertex is inactive<\/tt>.  The paper notes that inactive<\/tt> vertices can be reactivated, although it is a little vague on when this happens.<\/p>\n
Pregel's synchronous nature makes Pregel programs easy to think about. Although updates are done at many vertices in any single superstep, it doesn't matter in what order the updates are done, or if they're done in parallel, because the update at any specific vertex doesn't affect the result of updates at other vertices.  That means there's no possibility of race conditions arising.<\/p>\n
Single-machine Pregel library<\/h3>\n
I'll now describe a toy single-machine Pregel library, written in Python (v 2.6).  The main additional feature beyond the description of Pregel given above is that this library partitions the vertices and assigns the different parts of the partition to workers<\/em>, which in this implementation are separate Python threads.  As we'll see below, on a cluster this idea is extended so the partitioned vertices aren't just assigned to different threads, but may be assigned to different machines.  Here's the code (see also GitHub<\/a>):<\/p>\n
\r\n\"\"\"pregel.py is a python 2.6 module implementing a toy single-machine\r\nversion of Google's Pregel system for large-scale graph processing.\"\"\"\r\n\r\nimport collections\r\nimport threading\r\n\r\nclass Vertex():\r\n\r\n    def __init__(self,id,value,out_vertices):\r\n        # This is mostly self-explanatory, but has a few quirks:\r\n        #\r\n        # self.id is included mainly because it's described in the\r\n        # Pregel paper.  It is used briefly in the pagerank example,\r\n        # but not in any essential way, and I was tempted to omit it.\r\n        #\r\n        # Each vertex stores the current superstep number in\r\n        # self.superstep.  It's arguably not wise to store many copies\r\n        # of global state in instance variables, but Pregel's\r\n        # synchronous nature lets us get away with it.\r\n        self.id = id \r\n        self.value = value\r\n        self.out_vertices = out_vertices\r\n        self.incoming_messages = []\r\n        self.outgoing_messages = []\r\n        self.active = True\r\n        self.superstep = 0\r\n   \r\nclass Pregel():\r\n\r\n    def __init__(self,vertices,num_workers):\r\n        self.vertices = vertices\r\n        self.num_workers = num_workers\r\n\r\n    def run(self):\r\n        \"\"\"Runs the Pregel instance.\"\"\"\r\n        self.partition = self.partition_vertices()\r\n        while self.check_active():\r\n            self.superstep()\r\n            self.redistribute_messages()\r\n\r\n    def partition_vertices(self):\r\n        \"\"\"Returns a dict with keys 0,...,self.num_workers-1\r\n        representing the worker threads.  The corresponding values are\r\n        lists of vertices assigned to that worker.\"\"\"\r\n        partition = collections.defaultdict(list)\r\n        for vertex in self.vertices:\r\n            partition[self.worker(vertex)].append(vertex)\r\n        return partition\r\n\r\n    def worker(self,vertex):\r\n        \"\"\"Returns the id of the worker that vertex is assigned to.\"\"\"\r\n        return hash(vertex) % self.num_workers\r\n\r\n    def superstep(self):\r\n        \"\"\"Completes a single superstep.  \r\n\r\n        Note that in this implementation, worker threads are spawned,\r\n        and then destroyed during each superstep.  This creation and\r\n        destruction causes some overhead, and it would be better to\r\n        make the workers persistent, and to use a locking mechanism to\r\n        synchronize.  The Pregel paper suggests that this is how\r\n        Google's Pregel implementation works.\"\"\"\r\n        workers = []\r\n        for vertex_list in self.partition.values():\r\n            worker = Worker(vertex_list)\r\n            workers.append(worker)\r\n            worker.start()\r\n        for worker in workers:\r\n            worker.join()\r\n\r\n    def redistribute_messages(self):\r\n        \"\"\"Updates the message lists for all vertices.\"\"\"\r\n        for vertex in self.vertices:\r\n            vertex.superstep +=1\r\n            vertex.incoming_messages = []\r\n        for vertex in self.vertices:\r\n            for (receiving_vertix,message) in vertex.outgoing_messages:\r\n                receiving_vertix.incoming_messages.append((vertex,message))\r\n\r\n    def check_active(self):\r\n        \"\"\"Returns True if there are any active vertices, and False\r\n        otherwise.\"\"\"\r\n        return any([vertex.active for vertex in self.vertices])\r\n\r\nclass Worker(threading.Thread):\r\n\r\n    def __init__(self,vertices):\r\n        threading.Thread.__init__(self)\r\n        self.vertices = vertices\r\n\r\n    def run(self):\r\n        self.superstep()\r\n\r\n    def superstep(self):\r\n        \"\"\"Completes a single superstep for all the vertices in\r\n        self.\"\"\"\r\n        for vertex in self.vertices:\r\n            if vertex.active:\r\n                vertex.update()\r\n<\/pre>\n
Here's the Python code for a computation of PageRank using both the Pregel library just given and, as a test, a more conventional matrix-based approach.  You should not worry too much about the test code (at least initially), and concentrate on the bits related to Pregel.<\/p>\n
\r\n\"\"\"pagerank.py illustrates how to use the pregel.py library, and tests\r\nthat the library works.\r\n\r\nIt illustrates pregel.py by computing the PageRank for a randomly\r\nchosen 10-vertex web graph.\r\n\r\nIt tests pregel.py by computing the PageRank for the same graph in a\r\ndifferent, more conventional way, and showing that the two outputs are\r\nnear-identical.\"\"\"\r\n\r\nfrom pregel import *\r\n\r\n# The next two imports are only needed for the test.  \r\nfrom numpy import * \r\nimport random\r\n\r\nnum_workers = 4\r\nnum_vertices = 10\r\n\r\ndef main():\r\n    vertices = [PageRankVertex(j,1.0\/num_vertices,[]) \r\n                for j in range(num_vertices)]\r\n    create_edges(vertices)\r\n    pr_test = pagerank_test(vertices)\r\n    print \"Test computation of pagerank:\\n%s\" % pr_test\r\n    pr_pregel = pagerank_pregel(vertices)\r\n    print \"Pregel computation of pagerank:\\n%s\" % pr_pregel\r\n    diff = pr_pregel-pr_test\r\n    print \"Difference between the two pagerank vectors:\\n%s\" % diff\r\n    print \"The norm of the difference is: %s\" % linalg.norm(diff)\r\n\r\ndef create_edges(vertices):\r\n    \"\"\"Generates 4 randomly chosen outgoing edges from each vertex in\r\n    vertices.\"\"\"\r\n    for vertex in vertices:\r\n        vertex.out_vertices = random.sample(vertices,4)\r\n\r\ndef pagerank_test(vertices):\r\n    \"\"\"Computes the pagerank vector associated to vertices, using a\r\n    standard matrix-theoretic approach to computing pagerank.  This is\r\n    used as a basis for comparison.\"\"\"\r\n    I = mat(eye(num_vertices))\r\n    G = zeros((num_vertices,num_vertices))\r\n    for vertex in vertices:\r\n        num_out_vertices = len(vertex.out_vertices)\r\n        for out_vertex in vertex.out_vertices:\r\n            G[out_vertex.id,vertex.id] = 1.0\/num_out_vertices\r\n    P = (1.0\/num_vertices)*mat(ones((num_vertices,1)))\r\n    return 0.15*((I-0.85*G).I)*P\r\n\r\ndef pagerank_pregel(vertices):\r\n    \"\"\"Computes the pagerank vector associated to vertices, using\r\n    Pregel.\"\"\"\r\n    p = Pregel(vertices,num_workers)\r\n    p.run()\r\n    return mat([vertex.value for vertex in p.vertices]).transpose()\r\n\r\nclass PageRankVertex(Vertex):\r\n\r\n    def update(self):\r\n        # This routine has a bug when there are pages with no outgoing\r\n        # links (never the case for our tests).  This problem can be\r\n        # solved by introducing Aggregators into the Pregel framework,\r\n        # but as an initial demonstration this works fine.\r\n        if self.superstep < 50:\r\n            self.value = 0.15 \/ num_vertices + 0.85*sum(\r\n                [pagerank for (vertex,pagerank) in self.incoming_messages])\r\n            outgoing_pagerank = self.value \/ len(self.out_vertices)\r\n            self.outgoing_messages = [(vertex,outgoing_pagerank) \r\n                                      for vertex in self.out_vertices]\r\n        else:\r\n            self.active = False\r\n\r\nif __name__ == \"__main__\":\r\n    main()\r\n<\/pre>\n
You might observe that in the PageRank example the matrix-based code is no more complex than the Pregel-based code.  So why bother with Pregel?  What's different is that the Pregel code can, in principle, be automatically scaled on a cluster.  And that's not so easy for the matrix-based approach.<\/p>\n
The PageRank example also illustrates another point made in the Pregel paper.  The authors note that users of Pregel at Google find it easy to program using Pregel once they begin to \"think like a vertex\". In a sense, the update<\/tt> method for instances of PageRankVertex<\/tt> is simply the vertex asking itself \"What should I do in this superstep?\"<\/p>\n
There is, incidentally, a slight bug in the PageRank program.  The bug arises because of a quirk in PageRank.  Sometimes the random websurfer may come to a webpage that has no outgoing links.  When that happens, they can't simply choose a random outgoing link.  Instead, the PageRank algorithm is modified in this instance so that they always<\/em> teleport to a page chosen completely at random.<\/p>\n
A standard way of modelling this is to model a page with no outgoing links as a page that is linked to every single other page.  It sounds counterintuitive, but it works - if it is connected to every single other page, then a random websurfer will, indeed, teleport to a completely random page.<\/p>\n
Unfortunately, while this modification is fine in principle, it's not so good for Pregel.  When Pregel is put on a cluster, it requires a lot of network overhead to send messages to large numbers of vertices. But we'll see below that there's a way of modifying Pregel so it can deal with this.<\/p>\n
Problems<\/h3>\n