[SPARK-40745][MLLIB] Reduce the shuffle size of ALS in .mllib

[zhengruifeng]

What changes were proposed in this pull request?

Reduce the shuffle size of ALS by using Array[V] instead of BoundedPriorityQueue[V] in ser/deser
this is a corresponding change of #37918 on the .mllib side

Why are the changes needed?

Reduce the shuffle size of ALS

Does this PR introduce any user-facing change?

No

How was this patch tested?

existing UT

[zhengruifeng]

take RankingMetricsExample for example:

import org.apache.spark.mllib.evaluation.{RankingMetrics, RegressionMetrics}
import org.apache.spark.mllib.recommendation.{ALS, Rating}


val ratings = spark.read.textFile("data/mllib/sample_movielens_data.txt").rdd.map { line =>
    val fields = line.split("::")
    Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble - 2.5)
}.cache()

// Map ratings to 1 or 0, 1 indicating a movie that should be recommended
val binarizedRatings = ratings.map(r => Rating(r.user, r.product, if (r.rating > 0) 1.0 else 0.0)).cache()


val numIterations = 1
val rank = 10
val lambda = 0.01
val model = ALS.train(ratings, rank, numIterations, lambda)

val userRecommended = model.recommendProductsForUsers(10)

userRecommended.count()

before:

after:

the shuffle size was reduced from 310.1 KiB to 110.9 KiB

[zhengruifeng]

cc @srowen @WeichenXu123

[srowen]

Seems fine. Do you have one sentence about what the optimization is here - why is it smaller?

[zhengruifeng]

@srowen the new implementation uses Array[V] instead of BoundedPriorityQueue[V] in ser/deser

[srowen]

Merged to master