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
Closed

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

Changes from all commits
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
15 changes: 4 additions & 11 deletions 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 @@ -219,11 +219,6 @@ class KMeans private (
data: RDD[(Vector, Double)],
instr: Option[Instrumentation]): KMeansModel = {

if (data.getStorageLevel == StorageLevel.NONE) {
logWarning("The input data is not directly cached, which may hurt performance if its"
+ " parent RDDs are also uncached.")
}

// Compute squared norms and cache them.
val norms = data.map { case (v, _) =>
Vectors.norm(v, 2.0)
Expand All @@ -232,15 +227,13 @@ class KMeans private (
val zippedData = data.zip(norms).map { case ((v, w), norm) =>
(new VectorWithNorm(v, norm), w)
}
zippedData.persist(StorageLevel.MEMORY_AND_DISK)
val model = runAlgorithmWithWeight(zippedData, instr)
zippedData.unpersist()

// Warn at the end of the run as well, for increased visibility.
if (data.getStorageLevel == StorageLevel.NONE) {
Copy link

@yeyuqiang yeyuqiang Aug 12, 2020

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)
}

logWarning("The input data was not directly cached, which may hurt performance if its"
+ " parent RDDs are also uncached.")
zippedData.persist(StorageLevel.MEMORY_AND_DISK)
}
Copy link
Member

@srowen srowen Dec 30, 2019

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.

Copy link
Contributor Author

@amanomer amanomer Dec 30, 2019

Choose a reason for hiding this comment

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

Done

Copy link
Contributor

@zhengruifeng zhengruifeng Dec 31, 2019

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()
}

Copy link
Contributor Author

@amanomer amanomer Dec 31, 2019 •
edited
Loading

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?

Copy link
Member

@srowen srowen Dec 31, 2019

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()

model
}

Expand Down