Skip to content

[SPARK-20930][ML] Destroy broadcasted centers after computing cost in KMeans #18152

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
zhengruifeng wants to merge 4 commits into from
Closed

[SPARK-20930][ML] Destroy broadcasted centers after computing cost in KMeans #18152

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e49fbd3
create pr
zhengruifeng May 31, 2017
c390ab2
create pr
zhengruifeng May 31, 2017
9280726
fix gd & lda
zhengruifeng Jun 1, 2017
3fd52a8
del unused broadcasted object
zhengruifeng Jun 5, 2017
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: 4 additions & 1 deletion mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeansModel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,10 @@ class KMeansModel @Since("1.1.0") (@Since("1.0.0") val clusterCenters: Array[Vec
@Since("0.8.0")
def computeCost(data: RDD[Vector]): Double = {
val bcCentersWithNorm = data.context.broadcast(clusterCentersWithNorm)
data.map(p => KMeans.pointCost(bcCentersWithNorm.value, new VectorWithNorm(p))).sum()
val cost = data
.map(p => KMeans.pointCost(bcCentersWithNorm.value, new VectorWithNorm(p))).sum()
bcCentersWithNorm.destroy(blocking = false)
cost
}


Expand Down
4 changes: 2 additions & 2 deletions mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAModel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,7 @@ class LocalLDAModel private[spark] (

docBound
}.sum()
ElogbetaBc.destroy(blocking = false)
Copy link
Member

@srowen srowen Jun 1, 2017

Choose a reason for hiding this comment

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

Looks good; how about in the getTopicDistributionMethod method too? I actually think that broadcast is pointless.

Copy link
Contributor Author

@zhengruifeng zhengruifeng Jun 2, 2017

Choose a reason for hiding this comment

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

getTopicDistributionMethod method is only used in ml.clustering.LDA#transform, so I think this maybe useful if the model size is large.

Copy link
Member

@srowen srowen Jun 2, 2017

Choose a reason for hiding this comment

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

@zhengruifeng but the broadcast isn't actually used. Its .value is called, locally, not from a distributed method.

Copy link
Contributor Author

@zhengruifeng zhengruifeng Jun 5, 2017

Choose a reason for hiding this comment

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

@srowen You are right. I removed the unnecessary broadcasting.


// Bound component for prob(topic-term distributions):
// E[log p(beta | eta) - log q(beta | lambda)]
Expand Down Expand Up @@ -372,7 +373,6 @@ class LocalLDAModel private[spark] (
*/
private[spark] def getTopicDistributionMethod(sc: SparkContext): Vector => Vector = {
val expElogbeta = exp(LDAUtils.dirichletExpectation(topicsMatrix.asBreeze.toDenseMatrix.t).t)
val expElogbetaBc = sc.broadcast(expElogbeta)
val docConcentrationBrz = this.docConcentration.asBreeze
val gammaShape = this.gammaShape
val k = this.k
Expand All @@ -383,7 +383,7 @@ class LocalLDAModel private[spark] (
} else {
val (gamma, _, _) = OnlineLDAOptimizer.variationalTopicInference(
termCounts,
expElogbetaBc.value,
expElogbeta,
docConcentrationBrz,
gammaShape,
k)
Expand Down
1 change: 1 addition & 0 deletions mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,7 @@ object GradientDescent extends Logging {
// c: (grad, loss, count)
(c1._1 += c2._1, c1._2 + c2._2, c1._3 + c2._3)
})
bcWeights.destroy(blocking = false)

if (miniBatchSize > 0) {
/**
Expand Down