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

[amanomer]

What changes were proposed in this pull request?

Check before caching zippedData (as suggested in #26483 (comment)).

Why are the changes needed?

If the data is already cached before calling run method of KMeans then zippedData.persist() will hurt the performance. Hence, persisting it conditionally.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Manually.

[amanomer]

cc @srowen @zhengruifeng

[srowen]

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]

Done

[zhengruifeng]

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]

what about caching norms if data is already cached?

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

[srowen]

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

[srowen]

Jenkins, test this please

[SparkQA]

Test build #115963 has finished for PR 27052 at commit 7af2785.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[amanomer]

cc @srowen @zhengruifeng

[srowen]

We can change this further, but this is an improvement and less of a change than anything else we'd do. I'll merge it.

[amanomer]

Thanks @srowen

[yeyuqiang]

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