Skip to content

[SPARK-14322] [MLlib] Use treeAggregate instead of reduce in OnlineLDAOptimizer #12106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-14322] [MLlib] Use treeAggregate instead of reduce in OnlineLDAOptimizer #12106

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
38cf0f3
change reduce to treeReduce for lda
hhbyyh Apr 1, 2016
42ae469
use treeaggregate
hhbyyh Apr 6, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,10 +451,11 @@ final class OnlineLDAOptimizer extends LDAOptimizer {
}
Iterator((stat, gammaPart))
}
val statsSum: BDM[Double] = stats.map(_._1).reduce(_ += _)
val statsSum: BDM[Double] = stats.map(_._1).treeAggregate(BDM.zeros[Double](k, vocabSize))(
Copy link
Contributor

@mengxr mengxr Apr 1, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't seem right because the first arg is modified in-place, which violates reduce contract. It should be an aggregate (or treeAggregate) instead.

Copy link
Member

@jkbradley jkbradley Apr 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Whoops, that's a long-standing bug... Perhaps we can just backport this PR.

Copy link
Contributor Author

@hhbyyh hhbyyh Apr 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I understand, this does not generate any computation error since it still gives the correct sum, right?

Copy link
Member

@jkbradley jkbradley Apr 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would expect so in general, but it could return corrupted results in case of a failure.

Copy link
Member

@jkbradley jkbradley Apr 4, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the line which caused the original failure, so using treeAggregate here should help.

Copy link
Contributor Author

@hhbyyh hhbyyh Apr 5, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm thinking .treeReduce(_ + _) is fine here. Internally it will transform it into treeAggregate. Let me know if I'm wrong.

Copy link
Member

@jkbradley jkbradley Apr 5, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be better to modify the first argument, which treeAggregate should support. (Actually I noticed treeAggregate does not say it supports it in the docs, but it should be OK to assume. I just created [https://issues.apache.org/jira/browse/SPARK-14408] for that.)

Copy link
Contributor Author

@hhbyyh hhbyyh Apr 6, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Get it. Thanks

_ += _, _ += _)
expElogbetaBc.unpersist()
val gammat: BDM[Double] = breeze.linalg.DenseMatrix.vertcat(
stats.map(_._2).reduce(_ ++ _).map(_.toDenseMatrix): _*)
stats.map(_._2).flatMap(list => list).collect().map(_.toDenseMatrix): _*)
val batchResult = statsSum :* expElogbeta.t

// Note that this is an optimization to avoid batch.count
Expand Down