[SPARK-24290][ML] add support for Array input for instrumentation.logNamedValue

## What changes were proposed in this pull request?

Extend instrumentation.logNamedValue to support Array input
change the logging for "clusterSizes" to new method

## How was this patch tested?

N/A

Please review http://spark.apache.org/contributing.html before opening a pull request.

Author: Lu WANG <[email protected]>

Closes #21347 from ludatabricks/SPARK-24290.

Author: lu-wang-dl

mllib/src/main/scala/org/apache/spark/ml/clustering/BisectingKMeans.scala

@@ -276,8 +276,7 @@ class BisectingKMeans @Since("2.0.0") (
     val summary = new BisectingKMeansSummary(
       model.transform(dataset), $(predictionCol), $(featuresCol), $(k))
     model.setSummary(Some(summary))
-    // TODO: need to extend logNamedValue to support Array
-    instr.logNamedValue("clusterSizes", summary.clusterSizes.mkString("[", ",", "]"))
+    instr.logNamedValue("clusterSizes", summary.clusterSizes)
     instr.logSuccess(model)
     model
   }

mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala

@@ -426,8 +426,7 @@ class GaussianMixture @Since("2.0.0") (
       $(predictionCol), $(probabilityCol), $(featuresCol), $(k), logLikelihood)
     model.setSummary(Some(summary))
     instr.logNamedValue("logLikelihood", logLikelihood)
-    // TODO: need to extend logNamedValue to support Array
-    instr.logNamedValue("clusterSizes", summary.clusterSizes.mkString("[", ",", "]"))
+    instr.logNamedValue("clusterSizes", summary.clusterSizes)
     instr.logSuccess(model)
     model
   }

mllib/src/main/scala/org/apache/spark/ml/clustering/KMeans.scala

@@ -359,8 +359,7 @@ class KMeans @Since("1.5.0") (
       model.transform(dataset), $(predictionCol), $(featuresCol), $(k))
 
     model.setSummary(Some(summary))
-    // TODO: need to extend logNamedValue to support Array
-    instr.logNamedValue("clusterSizes", summary.clusterSizes.mkString("[", ",", "]"))
+    instr.logNamedValue("clusterSizes", summary.clusterSizes)
     instr.logSuccess(model)
     if (handlePersistence) {
       instances.unpersist()

mllib/src/main/scala/org/apache/spark/ml/util/Instrumentation.scala

@@ -132,6 +132,19 @@ private[spark] class Instrumentation[E <: Estimator[_]] private (
     log(compact(render(name -> value)))
   }
 
+  def logNamedValue(name: String, value: Array[String]): Unit = {
+    log(compact(render(name -> compact(render(value.toSeq)))))
+  }
+
+  def logNamedValue(name: String, value: Array[Long]): Unit = {
+    log(compact(render(name -> compact(render(value.toSeq)))))
+  }
+
+  def logNamedValue(name: String, value: Array[Double]): Unit = {
+    log(compact(render(name -> compact(render(value.toSeq)))))
+  }
+
+
   /**
    * Logs the successful completion of the training session.
    */