Yeah, spark has very little overhead per partition, so generally more partitions is better.
On Mon, Dec 8, 2014 at 1:46 PM, Theodore Vasiloudis < [email protected]> wrote: > @Daniel > > Not an expert either, I'm just going by what I see performance-wise > currently. Our groupByKey implementation was more than an order of > magnitude slower than using the self join and then reduceByKey. > FTA: > > *"pairs on the same machine with the same key are combined (by using the > lamdba function passed into reduceByKey) before the data is shuffled. [...] > On the other hand, when calling groupByKey - all the key-value pairs are > shuffled around. This is a lot of unnessary data to being transferred over > the network."* > Maybe this is the reason that using the self-join performs better. > Hopefully someone with more knowledge can enlighten us. > > Back on the topic, using Koert's suggestion of creating an explicit > Partitioner and caching the result I've been able to get ~1.5x improvement > in runtime for large datasets. I'm using 100 partitions currently, I'll > check if > increasing this value improves performance. Decreasing the number of > partitions has a large negative effect on the runtime. > > On Mon, Dec 8, 2014 at 5:46 PM, Daniel Darabos < > [email protected]> wrote: > >> >> On Mon, Dec 8, 2014 at 5:26 PM, Theodore Vasiloudis < >> [email protected]> wrote: >> >>> @Daniel >>> It's true that the first map in your code is needed, i.e. mapping so >>> that dstID is the new RDD key. >>> >> >> You wrote groupByKey is "highly inefficient due to the need to shuffle >> all the data", but you seem to agree that the shuffle cannot be avoided. Both >> approaches cause 1 shuffle. >> >> I still don't see why you expect a speedup from doing this with a join. >> But don't let me discourage you or anything. I'm not an expert, just trying >> to learn. >> >> The self-join on the dstKey will then create all the pairs of incoming >>> edges (plus self-referential and duplicates that need to be filtered out). >>> >>> @Koert >>> Are there any guidelines about setting the number of partitions in >>> HashParitioner then? >>> >>> What I know about my data is that the distribution of indegree value >>> (number of incoming edges for a vertex) will be similar to a power law >>> <https://en.wikipedia.org/wiki/Power_law>, i.e. >>> there will be a small number of keys with a high number of incoming >>> edges, while most of the keys will >>> have incoming few edges. >>> >>> What is a good partitioning strategy for a self-join on an RDD with >>> unbalanced key distributions? >>> >>> >>> >>> On Mon, Dec 8, 2014 at 3:59 PM, Daniel Darabos < >>> [email protected]> wrote: >>> >>>> I do not see how you hope to generate all incoming edge pairs without >>>> repartitioning the data by dstID. You need to perform this shuffle for >>>> joining too. Otherwise two incoming edges could be in separate partitions >>>> and never meet. Am I missing something? >>>> >>>> On Mon, Dec 8, 2014 at 3:53 PM, Theodore Vasiloudis < >>>> [email protected]> wrote: >>>> >>>>> Using groupByKey was our first approach, and as noted in the docs is >>>>> highly inefficient due to the need to shuffle all the data. See >>>>> http://databricks.gitbooks.io/databricks-spark-knowledge-base/content/best_practices/prefer_reducebykey_over_groupbykey.html >>>>> >>>>> On Mon, Dec 8, 2014 at 3:47 PM, Daniel Darabos < >>>>> [email protected]> wrote: >>>>> >>>>>> Could you not use a groupByKey instead of the join? I mean something >>>>>> like this: >>>>>> >>>>>> val byDst = rdd.map { case (src, dst, w) => dst -> (src, w) } >>>>>> byDst.groupByKey.map { case (dst, edges) => >>>>>> for { >>>>>> (src1, w1) <- edges >>>>>> (src2, w2) <- edges >>>>>> } { >>>>>> ??? // Do something. >>>>>> } >>>>>> ??? // Return something. >>>>>> } >>>>>> >>>>>> On Mon, Dec 8, 2014 at 3:28 PM, Koert Kuipers <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> spark can do efficient joins if both RDDs have the same partitioner. >>>>>>> so in case of self join I would recommend to create an rdd that has >>>>>>> explicit partitioner and has been cached. >>>>>>> On Dec 8, 2014 8:52 AM, "Theodore Vasiloudis" < >>>>>>> [email protected]> wrote: >>>>>>> >>>>>>>> Hello all, >>>>>>>> >>>>>>>> I am working on a graph problem using vanilla Spark (not GraphX) >>>>>>>> and at some >>>>>>>> point I would like to do a >>>>>>>> self join on an edges RDD[(srcID, dstID, w)] on the dst key, in >>>>>>>> order to get >>>>>>>> all pairs of incoming edges. >>>>>>>> >>>>>>>> Since this is the performance bottleneck for my code, I was >>>>>>>> wondering if >>>>>>>> there any steps to take before >>>>>>>> performing the self-join in order to make it as efficient as >>>>>>>> possible. >>>>>>>> >>>>>>>> In the Learning Spark book >>>>>>>> < >>>>>>>> https://www.safaribooksonline.com/library/view/learning-spark/9781449359034/ch04.html >>>>>>>> > >>>>>>>> for example, in the "Data partitioning" section they recommend >>>>>>>> performing .partitionBy(new HashPartitioner(100)) on an RDD before >>>>>>>> joining >>>>>>>> it with another. >>>>>>>> >>>>>>>> Are there any guidelines for optimizing self-join performance? >>>>>>>> >>>>>>>> Regards, >>>>>>>> Theodore >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> -- >>>>>>>> View this message in context: >>>>>>>> http://apache-spark-user-list.1001560.n3.nabble.com/Efficient-self-joins-tp20576.html >>>>>>>> Sent from the Apache Spark User List mailing list archive at >>>>>>>> Nabble.com. >>>>>>>> >>>>>>>> >>>>>>>> --------------------------------------------------------------------- >>>>>>>> To unsubscribe, e-mail: [email protected] >>>>>>>> For additional commands, e-mail: [email protected] >>>>>>>> >>>>>>>> >>>>>> >>>>> >>>> >>> >> >
