Thanks, the `foldByKey` suggestion is very helpful, and we'll add more comment to explain the tricky part.  But I'm not quite sure about the 2nd part.  In a previous version that I didn't commit I employed the method you suggests.  Although this method does avoid a shuffle, it leaves all the global reduction and the time consuming logarithm computation to the driver.  Also, in the worst case, as many as CxDxN `Double`s will be transferred to the driver (N is the partition number).  I think we should leave as much heavy computation as possible to the cluster rather than the driver, that's why I chose to use `reduceByKey`.

In one word, if the input size is not very large, the method without any shuffle is indeed faster, but I think the current solution is more scalable.