apache/incubator-spark — pull request #407 by yinxusen (2014-01-13 16:57:24+00:00)
Title: BSP+ version of GradientDescent to speedup machine learning codes

Hi, @liancheng and I write a kind of BSP+ version of `GradientDescent`, to speedup the initial `GradientDescent`. We find that with a few modifications of `runMiniBatch`, we can achieve a better effeciency of both classification and regression in mllib without sacrificing accuracy.

Put it simply, we just break the sequential update in initial version of `GradientDescent` with mapPartitions in the function `runMiniBatch`. Actually we run some inner iterations in each partition, say, 20 times loop, and get inner weights. Then we use a bulk synchronous to collect weights from all executors and merge them together. After that we start another outer iteration. Indeed, the modified version has almost the same DAG as the initial `GradientDescent`, taking `LogisticRegressionWithSGD` as an example in the DAG pictures below.

[Initial version of `LogisticRegressionWithSGD`]

![123](https://f.cloud.github.com/assets/2637239/1901663/dbd44be0-7c67-11e3-8c44-800a10f6d92a.jpg)

[BSP+ version of `LogisticRegressionWithSGD`]

![234](https://f.cloud.github.com/assets/2637239/1901664/e5fea980-7c67-11e3-9e24-5c9978d94d02.jpg)

In order to test both the accuracy and efficiency, we setup some experiments. The dataset is a public dataset of web spam detection (the unigram one is used, actually, I write a simple function to transform the data format from SVM one to LR one.), which can be found here: [web spam dataset](http://www.csie.ntu.edu.tw/~cjlin/libsvmtools/datasets/binary.html). In total, there are 254 features in each data entry, and 35w data entries. The size is 382MB.

We use 15 nodes cluster to test it, with SPARK_MEM=8g, SPARK_WORK_MEMORY=10g, and 120 CPU cores.

The experiment results is depicted as follows, I will introduce it with the pictures.

First of all, we investigate the convergence ability of BSP+ LR in comparison with the initial LR. Red line in each pictures refers to as BSP+ version while blue one represents initial version in mllib. For BSP+, the number of inner iterations is 20, and the number of outer iterations is also 20. For initial version, the number of iterations is 20. In conclusion, BSP+ LR iterates 400 times while initial version run 20 times.

Here we find that the BSP+ version is converged at outer iteration 3~4, which means that it reaches a good result not more than 4*20 = 80 iterations. However, initial version can't catch up with BSP+ version when it reaches 20 times. Below we talk about the time consumption.

![02](https://f.cloud.github.com/assets/2637239/1901927/995bf9da-7c6b-11e3-8356-ab93e775527b.png)

We can see that the in each iteration, the time consumption of BSP+ is 1.6x~1.9x than initial version on average. But remember that the initial version iterates 20 times, and also can’t catch up with BSP+ version with 3 outer iterations. So we can conclude that the initial version takes 20*6*10^9 = 1.2*10^11, while the BSP+ version beats it with 3*1.2*10^10 = 3.6*10^10. This is near an order of magnitude speedup.

![00](https://f.cloud.github.com/assets/2637239/1901926/99321ff2-7c6b-11e3-9dd2-e33b81acf329.png)

Then I run the initial version for more iterations (say, 80 iterations), to see if it converges at the same level of loss or better. The red one is BSP+ (20 inner iteration, 20 outer iteration), while the blue one is initial version of LR, with 80 iterations. We can see that BSP+ gives a better answer much faster.

![03](https://f.cloud.github.com/assets/2637239/1901928/99861274-7c6b-11e3-84d4-f332a8b0fe0f.png)

I use 15 nodes as workers in Spark Cluster, but not all nodes are used when I run the experiments. I did not set the number of partitions above. The Spark Perf told me that there are about 4~5 executors used in each iteration. I talked with @liancheng, we found that when we use HDFS as data source, it will compute a partition number itself. But we can also set the minimal number of partitions.

On condition that we have 15 nodes in total, I set the minimal partitions to 2, 4, 8, 15, but the real number of partitions is 6, 6, 8, 15. Here is the time consumption in each iteration of every different partition numbers.

![04](https://f.cloud.github.com/assets/2637239/1901929/99b08a7c-7c6b-11e3-909c-e2a78b1c9815.png)

The picture below is the total time consumption of each number of partitions.

![05](https://f.cloud.github.com/assets/2637239/1901930/99b89a28-7c6b-11e3-9c2d-70af0bec7b44.png)

Some more interesting results: The time consumption is a convex curve, it has a inflection point. As depicted in the picture below, we can say that the program will be the fastest when partition number = 45. Here is a trade-off between single partition accuracy with time consumption, which means that we can’t decrease the time consumption through the increasing of degree of parallelism all the time.

![06](https://f.cloud.github.com/assets/2637239/1901931/99b89672-7c6b-11e3-8ec3-22a792549445.png)

**A brief summary of the proposal:** Sequential version of updating in optimization algorithms limit the ability of parallel machine learning. Breaking the sequentail chain accurately and graciously is the key to add the degree of parallelism. Actually, this work is motivated by Jeff Dean's [DistBelief](http://research.google.com/archive/large_deep_networks_nips2012.html) and Eric Xing's [SSP](http://petuum.org/research.html). Some of these works can be implemented on Spark gracefully.

**Additional words:** This is just a proposal, because we modified the interface of `runMiniBatch` function (change the numIteration into a pair of numOuterIteration and numInnerIteration), it can not keep compatibility with the original version of code. If you feel that it is OK to merge in, I will refine the code in accordance with your request.