[WIP] [MLLIB-28] An optimized GradientDescent implementation

[yinxusen]

New JIRA issue MLLIB-28 with this pull request bring a new implementation of GradientDescent named GradientDescentWithLocalUpdate. The GradientDescentWithLocalUpdate can outperform the original GradientDescent by about 1x ~ 4x without sacrificing accuracy, and can be easily adopted by most classification and regression algorithms in MLlib.

Parallelism of many ML algorithms are limited by the sequential updating process of optimization algorithms they use. However, by carefully breaking the sequential chain, the updating process can be parallelized. In theGradientDescentWithLocalUpdate , we split the iteration loop into multiple supersteps. Within each superstep, an inner loop that runs a local optimization process is introduced into each partition. During the local optimization, only local data points in the partition are involved. Since different partitions are processed in parallel, the local optimization process is natually parallelized. Then, at the end of each superstep, all the gradients and loss histories computed from each partition are collected and merged in a bulk synchronous manner.

Detailed experiments and results in the original pull request and comments.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/13229/

[liancheng]

Indentation error, should be 2 spaces.

[srowen]

Broad question: can this simply replace the existing implementation, if it's better? I'd suggest it is important to not let a bunch of different implementations proliferate, but unify them.

[yinxusen]

In fact, if we set the numInnerIteration = 1, which is the default setting, then the GradientDescentWithLocalUpdate is identical to GradientDescent. However, I think it is better if we have opportunity to add new updater easily. End-user may have the requirement, for adding new customized updater.

[liancheng]

Left some comments on minor issues like formatting. LGTM otherwise.

@srowen According to previous experiments, this implementation is indeed better than the original one, +1 for the replacement.

[liancheng]

@srowen Forgot to mention, @etrain's comment should be one of the reasons why this PR doesn't try to replace the original one. BTW, basically I'm not an ML guy, so please ignore me if I'm saying rubbish :)

[mengxr]

I'm not very familiar with how duplicate is implemented. Scala doc says "The implementation may allocate temporary storage for elements iterated by one iterator but not yet by the other." Is there a risk of running out of memory here?

[yinxusen]

Good question, I look into duplicate method just now. It uses a scala.collection.mutuable.Queue to mimic an iterator, and the elements iterated by one iterator but not yet by the other is stored there. I am shocked by that...

I have no idea of the memory cost by the Queue, but it seems the only way to duplicate an iterator. We have already tested before that the method is really fast than iterator.toArray. @liancheng Do you know about that?

[liancheng]

We're using Iterator.duplicate to iterate the dataset multiple times without calling Iterator.toArray. According to implementation of duplicate, it does consume more memory and generates more temporary objects. I didn't notice that before. But according to previous experiments, duplicate is much more GC-friendly than toArray. I think the reason is that the underlying implementation of the mutable.Queue used in duplicate is actually a mutable.LinkedList, which doesn't require large amount of continuous memory, and thus may trigger full-GC less frequently.

If my guess is right, instead of using duplicate maybe we can simply call Iterator.toList to reduce full-GC frequency?

[mengxr]

@yinxusen I think this approach will certainly run OOM if data is too big to fit into memory. You can set a small executor memory and test some data without caching.

[yinxusen]

@mengxr, I absolutely agree with you. I am trying another way now, and will have a test result tomorrow.

[yinxusen]

I use the new method to enlarge local update. Test on SVM and LogisticRegression looks as good as the first version, without the worry of OOM. This method can get better result in shorter time, especially when the dataset is too large to cache in memory.

I think this method is much more like the method provided here in section 3. I'm not mentioned that it is a better way, but the original GradientDescent is somewhat like an elephant pulling a small carriage.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

One or more automated tests failed
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/13322/

[yinxusen]

I have test the original/1-version/2-version LR and SVM, here is the result:

(Note that original version runs 100 iterations, while the other two run 10 iterations with 10 local iterations.)

Type	Time	Last 10 losses
original LR	89.348	0.638 - 0.636
1-version LR	13.094	1.769 - 0.636
2-version LR	10.747	1.618 - 0.631
original SVM	88.708	0.947 - 0.947
1-version SVM	13.062	2.127 - 0.694
2-version SVM	10.829	1.943 - 0.691

There are 3 updaters : L1-updater, L2-updater, and simple updater, and 3 gradients: Logistic, Square and Hinge. SVM uses Hinge+L2, LR uses Logistic+simple, Lasso uses Square+L1.

But I encounter some difficulties in Lasso, I am still trying to fix them.

[AmplabJenkins]

Can one of the admins verify this patch?

[yinxusen]

I rewrite the 2 versions of GradientDescent with Vector instead of Array. Lasso is easy to test now thanks for @mengxr 's refactoring of code.

I run the test on a single node, in local mode. Note that original version runs 100 iterations, while the other two run 10 iterations with 10 local iterations.

latest update:

Type	Time	Last 10 losses
original LR	346	0.6444 - 0.6430
1-version LR	45	0.7082-0.6773
2-version LR	37	0.7070-0.6817
original SVM	338	0.9468 - 0.9468
1-version SVM	46	0.7861 - 0.7803
2-version SVM	34	0.7875 - 0.7829
original Lasso	320	0.6063 - 0.6063
1-version Lasso	39	0.6131 - 0.2062
2-version Lasso	32	0.6062 - 0.2104

1-version is not good due to the reuse of Iterator, which inherently store all elements in a queue and will cause OOM if the data entry in a partition is large enough. 2-version is better, but due to the tiny-batch property, 2-version GradientDescent's convergence ability is slightly lower than the 1-version. There is a trade-off between hardware efficiency and statistical efficiency.

I port my code into an independent git repo so as to do experiments more easily, I'll move them back here recently.

[yinxusen]

I'd like to close the PR, for the offline discussion with @mengxr . The code will be stay in my github repo, for those who still interested in it.