{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/3866","labels_url":"https://api.github.com/repos/apache/spark/issues/3866/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/3866/comments","events_url":"https://api.github.com/repos/apache/spark/issues/3866/events","html_url":"https://github.com/apache/spark/pull/3866","id":53211308,"number":3866,"title":"[SPARK-5036][Graphx]Better support sending partial messages in Pregel API","user":{"login":"shijinkui","id":648508,"avatar_url":"https://avatars.githubusercontent.com/u/648508?v=3","gravatar_id":"","url":"https://api.github.com/users/shijinkui","html_url":"https://github.com/shijinkui","followers_url":"https://api.github.com/users/shijinkui/followers","following_url":"https://api.github.com/users/shijinkui/following{/other_user}","gists_url":"https://api.github.com/users/shijinkui/gists{/gist_id}","starred_url":"https://api.github.com/users/shijinkui/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shijinkui/subscriptions","organizations_url":"https://api.github.com/users/shijinkui/orgs","repos_url":"https://api.github.com/users/shijinkui/repos","events_url":"https://api.github.com/users/shijinkui/events{/privacy}","received_events_url":"https://api.github.com/users/shijinkui/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":27,"created_at":"2015-01-01T02:11:28Z","updated_at":"2015-04-27T18:22:55Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/3866","html_url":"https://github.com/apache/spark/pull/3866","diff_url":"https://github.com/apache/spark/pull/3866.diff","patch_url":"https://github.com/apache/spark/pull/3866.patch"},"body":"# Better support sending partial messages in Pregel API\r\n\r\n###\t1. the reqirement\r\n\r\nIn many iterative graph algorithms, only a part of the vertexes (we call them ActiveVertexes) need to send messages to their neighbours in each iteration. In many cases, ActiveVertexes are the vertexes that their attributes do not change between the previous and current iteration. To implement this requirement, we can use Pregel API + a flag (e.g., `bool isAttrChanged`) in each vertex's attribute. \r\n\r\nHowever, after `aggregateMessage` or `mapReduceTriplets` of each iteration, we need to reset this flag to the init value in every vertex, which needs a heavy `joinVertices`. \r\n\r\nWe find a more efficient way to meet this requirement and want to discuss it here.\r\n\r\n\r\nLook at a simple example as follows:\r\n\r\nIn i-th iteartion, the previous attribute of each vertex is `Attr` and the newly computed attribute is `NewAttr`:\r\n\r\n|VID| Attr| NewAttr| Neighbours|\r\n|:----|:-----|:----|:------|\r\n| 1 | 4| 5| 2, 3 |\r\n| 2 | 3| 2| 1, 4 |\r\n| 3 | 2| 2| 1, 4 |\r\n| 4|  3| 4| 1, 2, 3 |\r\n\r\nOur requirement is that: \r\n\r\n1.\tSet each vertex's `Attr` to be `NewAttr` in i-th iteration    \r\n2.\tFor each vertex whose `Attr!=NewAttr`, send message to its neighbours in the next iteration's `aggregateMessage`.\r\n\r\n\r\nWe found it is hard to implement this requirment using current Pregel API efficiently. The reason is that we not only need to perform `pregel()` to  compute the `NewAttr`  (2) but also need to perform `outJoin()` to satisfy (1).\r\n\r\nA simple idea is to keep a `isAttrChanged:Boolean` (solution 1)  or `flag:Int` (solution 2) in each vertex's attribute.\r\n\r\n###\t2. two solution  \r\n-----------\r\n\r\n####\t2.1 solution 1: label and reset `isAttrChanged:Boolean` of Vertex Attr\r\n\r\n![s1](https://cloud.githubusercontent.com/assets/648508/5591527/3911977a-919f-11e4-9147-37cb3d2f940c.jpg)\r\n\r\n1. init message by `aggregateMessage`\r\n\tit return a messageRDD\r\n2. `innerJoin`\r\n\tcompute the messages on the received vertex, return a new VertexRDD which have the computed value by customed logic function `vprog`, set `isAttrChanged = true`\r\n3. `outerJoinVertices`\r\n\tupdate the changed vertex to the whole graph. now the graph is new.\r\n4. `aggregateMessage`. it return a messageRDD\r\n5. `joinVertices`  reset erery `isAttrChanged` of Vertex attr to false\r\n\r\n\t```\r\n\t//\there reset the isAttrChanged to false\r\n\tg = updateG.joinVertices(updateG.vertices) {\r\n      \t(vid, oriVertex, updateGVertex) => updateGVertex.reset()\r\n    \t}\r\n   ```\r\n   here need to reset the vertex attribute object's variable as false\r\n\r\nif don't reset the `isAttrChanged`, it will send message next iteration directly.\r\n\r\n**result:**  \r\n\r\n*\tEdge: 890041895 \r\n*\tVertex: 181640208\r\n*\tIterate: 150 times\r\n*\tCost total: 8.4h\r\n*\tcan't run until the 0 message \r\n\r\n\r\n####\tsolution 2. color vertex. (Choose this)\r\n\r\n![s2](https://cloud.githubusercontent.com/assets/648508/5591530/434bafc8-919f-11e4-8ac7-44a8a2314cb5.jpg)\r\n\r\niterate process:\r\n\r\n1. innerJoin \r\n  `vprog` using as a partial function, looks like `vprog(curIter, _: VertexId, _: VD, _: A)`\r\n  ` i = i + 1; val curIter = i`. \r\n  in `vprog`, user can fetch `curIter` and assign to `falg`.\r\n2. outerJoinVertices\r\n\t`graph = graph.outerJoinVertices(changedVerts) { (vid, old, newOpt) => newOpt.getOrElse(old)}.cache()`\r\n3. aggregateMessages     \r\n\tsendMsg is partial function, looks like `sendMsg(curIter, _: EdgeContext[VD, ED, A]`    \t\r\n\t**in `sendMsg`, compare `curIter` with `flag`, determine whether sending message**\r\n\r\n####\tresult\r\n\r\nraw data       from\r\n\r\n*\tvertex: 181640208\r\n*\tedge: 890041895\r\n\r\n\r\n|  | iteration average cost | 150 iteration cost | 420 iteration cost | \r\n| ------------ | ------------- | ------------ | ------------ |\r\n|  solution 1 | 188m | 7.8h | cannot finish  |\r\n|  solution 2 | 24m | 1.2h   | 3.1h | \r\n| compare  | 7x  | 6.5x  | finished in 3.1 |\r\n\r\n    \r\n##\t\tthe end\r\n    \r\ni think the second solution(Pregel + a flag) is better.    \r\nthis can really support the iterative graph algorithms which only part of the vertexes send messages to their neighbours in each iteration."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/96769896","html_url":"https://github.com/apache/spark/pull/3866#issuecomment-96769896","issue_url":"https://api.github.com/repos/apache/spark/issues/3866","id":96769896,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2015-04-27T18:22:55Z","updated_at":"2015-04-27T18:22:55Z","body":"Can one of the admins verify this patch?"}}