charles/scala_parallelism_concurrency
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Test 2

Browse Source
This commit is contained in:
Kali Live user 2020-03-15 12:02:25 +00:00
parent 9a3514a468
commit 16b6696555
1 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/main/scala/kmeans/KMeans.scala
View File

@ -45,10 +45,9 @@ class KMeans extends KMeansInterface {
} }
def classify(points: ParSeq[Point], means: ParSeq[Point]): ParMap[Point, ParSeq[Point]] = { def classify(points: ParSeq[Point], means: ParSeq[Point]): ParMap[Point, ParSeq[Point]] = {
val meansWithPoints = points.groupBy( p => findClosest(p, means) ) val pointsMeanMap = points.par.groupBy(findClosest(_, means))
means // So iterate over means get (empty) list and return map
.map(m => (m, if (meansWithPoints.contains(m)) List() ++ meansWithPoints(m) else List())) means.par.map(mean => mean -> pointsMeanMap.getOrElse(mean, ParSeq())).toMap
.toMap
} }
def findAverage(oldMean: Point, points: ParSeq[Point]): Point = if (points.isEmpty) oldMean else { def findAverage(oldMean: Point, points: ParSeq[Point]): Point = if (points.isEmpty) oldMean else {
@ -64,20 +63,21 @@ class KMeans extends KMeansInterface {
} }
def update(classified: ParMap[Point, ParSeq[Point]], oldMeans: ParSeq[Point]): ParSeq[Point] = { def update(classified: ParMap[Point, ParSeq[Point]], oldMeans: ParSeq[Point]): ParSeq[Point] = {
oldMeans.map( mean => findAverage(mean, classified(mean)) ) oldMeans.par.map(oldMean => findAverage(oldMean, classified(oldMean)))
} }
def converged(eta: Double, oldMeans: ParSeq[Point], newMeans: ParSeq[Point]): Boolean = { def converged(eta: Double, oldMeans: ParSeq[Point], newMeans: ParSeq[Point]): Boolean = {
oldMeans (oldMeans zip newMeans).forall{
.zip(newMeans) case (oldMean, newMean) => oldMean.squareDistance(newMean) <= eta
.map(p => p._1.squareDistance(p._2) <= eta) }
.forall( t => t )
} }
@tailrec @tailrec
final def kMeans(points: ParSeq[Point], means: ParSeq[Point], eta: Double): ParSeq[Point] = { final def kMeans(points: ParSeq[Point], means: ParSeq[Point], eta: Double): ParSeq[Point] = {
val freshMints = update(classify(points, means), means) val classified = classify(points, means)
if (!converged(eta)(means, freshMints)) kMeans(points, freshMints, eta) else freshMints // your implementation need to be tail recursive val newMeans = update(classified, means)
if (!converged(eta, means, newMeans)) kMeans(points, newMeans, eta) else newMeans
} }
} }