Hi Imran,
I will say your explanation is extremely helpful J I tested some ideas according to your explanation and it make perfect sense to me. I modify my code to use cogroup+mapValues instead of union+reduceByKey to preserve the partition, which gives me more than 100% performance gain (for the loop part). Thanks a lot! And I am curious will there any easy way for me to get a detail DAG execution plan description without running the code? Just as explain command in pig or sql? Shuai From: Imran Rashid [mailto:iras...@cloudera.com] Sent: Monday, February 23, 2015 6:00 PM To: Shuai Zheng Cc: Shao, Saisai; user@spark.apache.org Subject: Re: Union and reduceByKey will trigger shuffle even same partition? I think you're getting tripped up lazy evaluation and the way stage boundaries work (admittedly its pretty confusing in this case). It is true that up until recently, if you unioned two RDDs with the same partitioner, the result did not have the same partitioner. But that was just fixed here: https://github.com/apache/spark/pull/4629 That does mean that after you update ranks, it will no longer have a partitioner, which will effect the join on your second iteration here: val contributions = links.join(ranks).flatMap But, I think most of the shuffles you are pointing to are a different issue. I may be belaboring something you already know, but I think this is easily confusing. I think the first thing is understanding where you get stage boundaries, and how they are named. Each shuffle introduces a stage boundary. However, the stages get named by the last thing in a stage, which is not really what is always causing the shuffle. Eg., reduceByKey() causes a shuffle, but we don't see that in a stage name. Similarly, map() does not cause a shuffle, but we see a stage with that name. So, what do the stage boundaries we see actually correspond to? 1) map -- that is doing the shuffle write for the following groupByKey 2) groupByKey -- in addition to reading the shuffle output from your map, this is *also* doing the shuffle write for the next shuffle you introduce w/ partitionBy 3) union -- this is doing the shuffle reading from your partitionBy, and then all the work from there right up until the shuffle write for what is immediatley after union -- your reduceByKey. 4) lookup is an action, which is why that has another stage. a couple of things to note: (a) your join does not cause a shuffle, b/c both rdds share a partitioner (b) you have two shuffles from groupByKey followed by partitionBy -- you really probably want the 1 arg form of groupByKey(partitioner) hopefully this is helpful to understand how your stages & shuffles correspond to your code. Imran On Mon, Feb 23, 2015 at 3:35 PM, Shuai Zheng <szheng.c...@gmail.com> wrote: This also trigger an interesting question: how can I do this locally by code if I want. For example: I have RDD A and B, which has some partition, then if I want to join A to B, I might just want to do a mapper side join (although B itself might be big, but B’s local partition is known small enough put in memory), how can I access other RDD’s local partition in the mapParitition method? Is it anyway to do this in Spark? From: Shao, Saisai [mailto:saisai.s...@intel.com] Sent: Monday, February 23, 2015 3:13 PM To: Shuai Zheng Cc: user@spark.apache.org Subject: RE: Union and reduceByKey will trigger shuffle even same partition? If you call reduceByKey(), internally Spark will introduce a shuffle operations, not matter the data is already partitioned locally, Spark itself do not know the data is already well partitioned. So if you want to avoid Shuffle, you have to write the code explicitly to avoid this, from my understanding. You can call mapParitition to get a partition of data and reduce by key locally by your logic. Thanks Saisai From: Shuai Zheng [mailto:szheng.c...@gmail.com] Sent: Monday, February 23, 2015 12:00 PM To: user@spark.apache.org Subject: Union and reduceByKey will trigger shuffle even same partition? Hi All, I am running a simple page rank program, but it is slow. And I dig out part of reason is there is shuffle happen when I call an union action even both RDD share the same partition: Below is my test code in spark shell: import org.apache.spark.HashPartitioner sc.getConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") val beta = 0.8 val numOfPartition = 6 val links = sc.textFile("c:/Download/web-Google.txt").filter(!_.contains("#")).map(line=>{val part=line.split("\t"); (part(0).toInt,part(1).toInt)}).groupByKey.partitionBy(new HashPartitioner(numOfPartition)).persist var ranks = links.mapValues(_ => 1.0) var leakedMatrix = links.mapValues(_ => (1.0-beta)).persist for (i <- 1 until 2) { val contributions = links.join(ranks).flatMap { case (pageId, (links, rank)) => links.map(dest => (dest, rank / links.size * beta)) } ranks = contributions.union(leakedMatrix).reduceByKey(_ + _) } ranks.lookup(1) In above code, links will join ranks and should preserve the partition, and leakedMatrix also share the same partition, so I expect there is no shuffle happen on the contributions.union(leakedMatrix), also on the coming reduceByKey after that. But finally there is shuffle write for all steps, map, groupByKey, Union, partitionBy, etc. I expect there should only happen once on the shuffle then all should local operation, but the screen shows not, do I have any misunderstanding here?
