Skip to content

[SPARK-30390][MLLIB] Avoid double caching in mllib.KMeans#runWithWeights. #27052

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 3 commits into from
+4 −11
Closed

[SPARK-30390][MLLIB] Avoid double caching in mllib.KMeans#runWithWeights. #27052

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2e6f156
Initial commit
amanomer Dec 30, 2019
e7b1587
Remove warnings
amanomer Dec 30, 2019
7af2785
ScalaStyle Fix
amanomer Dec 30, 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
5 changes: 4 additions & 1 deletion mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,10 @@ class KMeans private (
val zippedData = data.zip(norms).map { case ((v, w), norm) =>
(new VectorWithNorm(v, norm), w)
}
zippedData.persist(StorageLevel.MEMORY_AND_DISK)

if (data.getStorageLevel == StorageLevel.NONE) {

@yeyuqiang yeyuqiang Aug 12, 2020

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Hi, I was testing spark kmeans. There should be an issue that no matter we persist the parent RDD, here the data.getStorageLevel will always be NONE due to the following operation, this will cause double caching.

def run(data: RDD[Vector]): KMeansModel = {
    val instances: RDD[(Vector, Double)] = data.map {
      case (point) => (point, 1.0)
    }
    runWithWeight(instances, None)
}

zippedData.persist(StorageLevel.MEMORY_AND_DISK)
}

@srowen srowen Dec 30, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I guess we can remove the two warnings in this method? it's not a big deal now if the source data is uncached.

@amanomer amanomer Dec 30, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Done

@zhengruifeng zhengruifeng Dec 31, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

what about caching norms if data is already cached? like this:

val handlePersistence = data.getStorageLevel == StorageLevel.NONE
val norms = ...
val zippedData = if (handlePersistence) {
   data.zip(norms).map { case ((v, w), norm) =>
      (new VectorWithNorm(v, norm), w)
    }.persist(StorageLevel.MEMORY_AND_DISK)
} else {
     norms.persist(StorageLevel.MEMORY_AND_DISK)
     data.zip(norms).map { case ((v, w), norm) =>
        (new VectorWithNorm(v, norm), w)
     }
}

...


if (handlePersistence) {
   zippedData.unpersist()
} else {
   norms.unpersist()
}

@amanomer amanomer Dec 31, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

what about caching norms if data is already cached?

Won't this lead to double caching problem which we are trying to avoid?

@srowen srowen Dec 31, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Yeah I thought that was your point. If zippedData were expensive, I'd agree that caching the intermediate values too is worthwhile, and we do that in some places. Here it's not, and the original behavior was to always cache internally, so I guess this is less of a change. This at least skips it where it can be inexpensively computed

val model = runAlgorithmWithWeight(zippedData, instr)
zippedData.unpersist()

Expand Down