Hey Reza,
Thanks for your response!
Your response clarifies some of my initial thoughts. However, what I don't
understand is how the depth of the tree is used to identify how many
intermediate reducers there will be, and how many partitions are sent to
the intermediate reducers. Could you provide some insight into this?
Thanks,
Raghav
On Thursday, June 4, 2015, Reza Zadeh <r...@databricks.com> wrote:
> In a regular reduce, all partitions have to send their reduced value to a
> single machine, and that machine can become a bottleneck.
>
> In a treeReduce, the partitions talk to each other in a logarithmic number
> of rounds. Imagine a binary tree that has all the partitions at its leaves
> and the root will contain the final reduced value. This way there is no
> single bottleneck machine.
>
> It remains to decide the number of children each node should have and how
> deep the tree should be, which is some of the logic in the method you
> pasted.
>
> On Wed, Jun 3, 2015 at 7:10 PM, raggy <raghav0110...@gmail.com
> <javascript:_e(%7B%7D,'cvml','raghav0110...@gmail.com');>> wrote:
>
>> I am trying to understand what the treeReduce function for an RDD does,
>> and
>> how it is different from the normal reduce function. My current
>> understanding is that treeReduce tries to split up the reduce into
>> multiple
>> steps. We do a partial reduce on different nodes, and then a final reduce
>> is
>> done to get the final result. Is this correct? If so, I guess what I am
>> curious about is, how does spark decide how many nodes will be on each
>> level, and how many partitions will be sent to a given node?
>>
>> The bulk of the implementation is within this function:
>>
>> partiallyReduced.treeAggregate(Option.empty[T])(op, op, depth)
>> .getOrElse(throw new UnsupportedOperationException("empty
>> collection"))
>>
>> The above function is expanded to
>>
>> val cleanSeqOp = context.clean(seqOp)
>> val cleanCombOp = context.clean(combOp)
>> val aggregatePartition =
>> (it: Iterator[T]) => it.aggregate(zeroValue)(cleanSeqOp,
>> cleanCombOp)
>> var partiallyAggregated = mapPartitions(it =>
>> Iterator(aggregatePartition(it)))
>> var numPartitions = partiallyAggregated.partitions.length
>> val scale = math.max(math.ceil(math.pow(numPartitions, 1.0 /
>> depth)).toInt, 2)
>> // If creating an extra level doesn't help reduce
>> // the wall-clock time, we stop tree aggregation.
>> while (numPartitions > scale + numPartitions / scale) {
>> numPartitions /= scale
>> val curNumPartitions = numPartitions
>> partiallyAggregated = partiallyAggregated.mapPartitionsWithIndex {
>> (i, iter) => iter.map((i % curNumPartitions, _))
>> }.reduceByKey(new HashPartitioner(curNumPartitions),
>> cleanCombOp).values
>> }
>> partiallyAggregated.reduce(cleanCombOp)
>>
>> I am completely lost about what is happening in this function. I would
>> greatly appreciate some sort of explanation.
>>
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-spark-user-list.1001560.n3.nabble.com/TreeReduce-Functionality-in-Spark-tp23147.html
>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
>> <javascript:_e(%7B%7D,'cvml','user-unsubscr...@spark.apache.org');>
>> For additional commands, e-mail: user-h...@spark.apache.org
>> <javascript:_e(%7B%7D,'cvml','user-h...@spark.apache.org');>
>>
>>
>
