Fix RobustKMeans divergence case mismatch and SparseKMeans L1 ClassCastException

- ClusteringOps: normalize all divergence lookups with toLowerCase, add aliases (se, is, logisticloss)
- ClusteringOps: align isValidDivergence and createUpdateStrategy with case-insensitive matching
- SparseKMeans.updateCenters: pattern match BregmanKernel for grad-based update; use MedianUpdateStrategy for L1 (component-wise median, not mean)

Author: derrickburns

src/main/scala/com/massivedatascience/clusterer/ml/SparseKMeans.scala

@@ -369,39 +369,62 @@ class SparseKMeans(override val uid: String)
       kernel: ClusteringKernel,
       numClusters: Int
   ): Array[Vector] = {
-    val bregmanKernel = kernel.asInstanceOf[BregmanKernel]
-    val bcKernel      = assigned.sparkSession.sparkContext.broadcast(bregmanKernel)
-
-    val gradUDF = udf { (features: Vector) =>
-      bcKernel.value.grad(features).toArray
-    }
+    val dim = assigned.select($(featuresCol)).head().getAs[Vector](0).size
 
-    val withGrad = assigned.withColumn("_grad", gradUDF(col($(featuresCol))))
+    kernel match {
+      case bk: BregmanKernel =>
+        // Use gradient-based update for Bregman divergences
+        val bcKernel = assigned.sparkSession.sparkContext.broadcast(bk)
 
-    val dim = assigned.select($(featuresCol)).head().getAs[Vector](0).size
+        val gradUDF = udf { (features: Vector) =>
+          bcKernel.value.grad(features).toArray
+        }
 
-    val aggregated = withGrad
-      .groupBy("_cluster")
-      .agg(
-        count("*").as("count"),
-        array((0 until dim).map(i => sum(element_at(col("_grad"), i + 1))): _*).as("grad_sum")
-      )
-      .collect()
+        val withGrad = assigned.withColumn("_grad", gradUDF(col($(featuresCol))))
+
+        val aggregated = withGrad
+          .groupBy("_cluster")
+          .agg(
+            count("*").as("count"),
+            array((0 until dim).map(i => sum(element_at(col("_grad"), i + 1))): _*).as("grad_sum")
+          )
+          .collect()
+
+        val centers = Array.fill(numClusters)(Vectors.zeros(dim))
+        aggregated.foreach { row =>
+          val clusterId = row.getInt(0)
+          if (clusterId >= 0 && clusterId < numClusters) {
+            val count   = row.getLong(1)
+            val gradSum = row.getSeq[Double](2).toArray
+            if (count > 0) {
+              val avgGrad = Vectors.dense(gradSum.map(_ / count))
+              centers(clusterId) = bcKernel.value.invGrad(avgGrad)
+            }
+          }
+        }
 
-    val centers = Array.fill(numClusters)(Vectors.zeros(dim))
-    aggregated.foreach { row =>
-      val clusterId = row.getInt(0)
-      if (clusterId >= 0 && clusterId < numClusters) {
-        val count   = row.getLong(1)
-        val gradSum = row.getSeq[Double](2).toArray
-        if (count > 0) {
-          val avgGrad = Vectors.dense(gradSum.map(_ / count))
-          centers(clusterId) = bcKernel.value.invGrad(avgGrad)
+        centers
+
+      case _ =>
+        // Non-Bregman kernels (e.g., L1): use component-wise median via
+        // MedianUpdateStrategy, which correctly minimizes L1 distance.
+        val updateStrategy = ClusteringOps.createUpdateStrategy("l1")
+        val renamed        = assigned.withColumnRenamed("_cluster", "cluster")
+        val medianCenters  = updateStrategy.update(
+          renamed,
+          $(featuresCol),
+          weightCol = None,
+          k = numClusters,
+          kernel
+        )
+        // MedianUpdateStrategy may return fewer centers (drops empty clusters).
+        // Pad back to numClusters with zeros.
+        val centers        = Array.fill(numClusters)(Vectors.zeros(dim))
+        medianCenters.zipWithIndex.foreach { case (c, i) =>
+          if (i < numClusters) centers(i) = Vectors.dense(c)
         }
-      }
+        centers
     }
-
-    centers
   }
 
   private def computeMovement(

src/main/scala/com/massivedatascience/clusterer/ml/df/ClusteringOps.scala

@@ -50,13 +50,16 @@ private[ml] object ClusteringOps extends Logging {
     "spherical"
   )
 
-  /** All supported divergence names and aliases. */
+  /** All supported divergence names and aliases (lowercased for case-insensitive matching). */
   private val validDivergenceNames: Set[String] = Set(
-    "squaredEuclidean",
+    "squaredeuclidean",
+    "se",
     "kl",
-    "itakuraSaito",
-    "generalizedI",
+    "itakurasaito",
+    "is",
+    "generalizedi",
     "logistic",
+    "logisticloss",
     "l1",
     "manhattan",
     "spherical",
@@ -75,15 +78,15 @@ private[ml] object ClusteringOps extends Logging {
     *   if divergence name is unknown
     */
   def createKernel(divergence: String, smoothing: Double = 1e-10): ClusteringKernel = {
-    divergence match {
-      case "squaredEuclidean"     => new SquaredEuclideanKernel()
-      case "kl"                   => new KLDivergenceKernel(smoothing)
-      case "itakuraSaito"         => new ItakuraSaitoKernel(smoothing)
-      case "generalizedI"         => new GeneralizedIDivergenceKernel(smoothing)
-      case "logistic"             => new LogisticLossKernel(smoothing)
-      case "l1" | "manhattan"     => new L1Kernel()
-      case "spherical" | "cosine" => new SphericalKernel()
-      case _                      =>
+    divergence.toLowerCase match {
+      case "squaredeuclidean" | "se"   => new SquaredEuclideanKernel()
+      case "kl"                        => new KLDivergenceKernel(smoothing)
+      case "itakurasaito" | "is"       => new ItakuraSaitoKernel(smoothing)
+      case "generalizedi"              => new GeneralizedIDivergenceKernel(smoothing)
+      case "logistic" | "logisticloss" => new LogisticLossKernel(smoothing)
+      case "l1" | "manhattan"          => new L1Kernel()
+      case "spherical" | "cosine"      => new SphericalKernel()
+      case _                           =>
         throw new IllegalArgumentException(
           s"Unknown divergence: '$divergence'. " +
             s"Valid options: ${supportedDivergences.mkString(", ")}"
@@ -122,7 +125,7 @@ private[ml] object ClusteringOps extends Logging {
     *   configured UpdateStrategy
     */
   def createUpdateStrategy(divergence: String): UpdateStrategy = {
-    divergence match {
+    divergence.toLowerCase match {
       case "l1" | "manhattan" => new MedianUpdateStrategy()
       case _                  => new GradMeanUDAFUpdate()
     }
@@ -181,5 +184,5 @@ private[ml] object ClusteringOps extends Logging {
   /** Check if a divergence name is valid.
     */
   def isValidDivergence(divergence: String): Boolean =
-    validDivergenceNames.contains(divergence)
+    validDivergenceNames.contains(divergence.toLowerCase)
 }