Skip to content

[SPARK-29818][MLLIB] Missing persist on RDD #26454

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
amanomer wants to merge 6 commits into from
Closed

[SPARK-29818][MLLIB] Missing persist on RDD #26454

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4acb797
Initial commit
amanomer Nov 9, 2019
fa920f7
Add persist for more files SPARK-29828, SPARK-29826
amanomer Nov 11, 2019
daad006
Updated based on review comments
amanomer Nov 13, 2019
cec29ff
Handled review comments
amanomer Nov 13, 2019
4aa39fc
Check before persisting
amanomer Nov 13, 2019
31c0fe7
Added unpersist
amanomer Nov 13, 2019
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
10 changes: 8 additions & 2 deletions mllib/src/main/scala/org/apache/spark/mllib/evaluation/BinaryClassificationMetrics.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.mllib.evaluation.binary._
import org.apache.spark.rdd.{RDD, UnionRDD}
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.storage.StorageLevel

/**
* Evaluator for binary classification.
Expand Down Expand Up @@ -165,13 +166,17 @@ class BinaryClassificationMetrics @Since("3.0.0") (
confusions: RDD[(Double, BinaryConfusionMatrix)]) = {
// Create a bin for each distinct score value, count weighted positives and
// negatives within each bin, and then sort by score values in descending order.
val counts = scoreLabelsWeight.combineByKey(
val binnedWeights = scoreLabelsWeight.combineByKey(
createCombiner = (labelAndWeight: (Double, Double)) =>
new BinaryLabelCounter(0.0, 0.0) += (labelAndWeight._1, labelAndWeight._2),
mergeValue = (c: BinaryLabelCounter, labelAndWeight: (Double, Double)) =>
c += (labelAndWeight._1, labelAndWeight._2),
mergeCombiners = (c1: BinaryLabelCounter, c2: BinaryLabelCounter) => c1 += c2
).sortByKey(ascending = false)
)
if (scoreLabelsWeight.getStorageLevel != StorageLevel.NONE) {
binnedWeights.persist()
Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

This still isn't unpersisted

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019

Choose a reason for hiding this comment

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

Do you mean it should be unpersisted after use?

Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

Yes otherwise the caller has no way to unpersist it until it's GCed

}
val counts = binnedWeights.sortByKey(ascending = false)
Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

Wait, hm, I don't understand this. You persist binnedWeights, but it is now only used once. Why? If anything it's binnedCounts that needs persisting. I'm still not clear if it makes enough difference to matter.

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019

Choose a reason for hiding this comment

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

binnedCounts is a child RDD of binnedWeights. And here one action sortByKey is performed on binnedWeights.

Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

Yes, but, why bother persisting binnedWeights? you recompute everything in between it and binnedCounts twice, when I think that would be the point, to avoid that.

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

I think binnedWeights is required to be persisted because more than one action is getting applied here.

binnedWeights
      |
      | sortByKey (action)
     V
counts
      |
      | count (action)
     V
binnedCounts (on which action collect is applied to compute agg)

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019

Choose a reason for hiding this comment

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

I might be wrong here. Kindly correct me @srowen

Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

caching helps where more than one action is performed on the same RDD. That's not the case here. Each of the first two has one thing executed on it. sortByKey is not an action, anyway.

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019

Choose a reason for hiding this comment

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

Oh, okay. One question here, will it be worth persisting counts since actions count and collect is applied directly on it ?

Copy link
Member

@srowen srowen Nov 13, 2019

Choose a reason for hiding this comment

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

Doesn't seem so. But that is the question I'd put to you in these cases - are you sure it makes a difference meaningful enough to overcome the overhead? I could imagine so here, just wondering if these are based on more investigation or benchmarking, vs just trying to persist lots of things.

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019

Choose a reason for hiding this comment

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

are you sure it makes a difference meaningful enough to overcome the overhead?

I think, no. Persisting count doesn't makes sense here. It will just be an overhead. Now I am getting clear picture of where to use persist.
Key learnings from this PR about persist.

  • persist introduce memory and CPU overheads.

  • So only important inputs (such as intermediate results, user data which is already cached, etc) should be persisted or RDD on which more than one action is performed.

  • Avoid using persist in loop.

  • Persist should be meaningful enough to overcome overheads.

Copy link
Contributor Author

@amanomer amanomer Nov 13, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

TYSM @srowen . Looking forward for more learning opportunities.


val binnedCounts =
// Only down-sample if bins is > 0
Expand Down Expand Up @@ -215,6 +220,7 @@ class BinaryClassificationMetrics @Since("3.0.0") (
val partitionwiseCumulativeCounts =
agg.scanLeft(new BinaryLabelCounter())((agg, c) => agg.clone() += c)
val totalCount = partitionwiseCumulativeCounts.last
binnedWeights.unpersist()
logInfo(s"Total counts: $totalCount")
val cumulativeCounts = binnedCounts.mapPartitionsWithIndex(
(index: Int, iter: Iterator[(Double, BinaryLabelCounter)]) => {
Expand Down