You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2025/02/08 00:47:00 UTC
(spark) branch branch-4.0 updated: [SPARK-50945][ML][PYTHON][CONNECT] Support Summarizer and SummaryBuilder on Connect
This is an automated email from the ASF dual-hosted git repository.
ruifengz pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 5d716f1049aa [SPARK-50945][ML][PYTHON][CONNECT] Support Summarizer and SummaryBuilder on Connect
5d716f1049aa is described below
commit 5d716f1049aae1d80df4fed050f213a97a058145
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Sat Feb 8 08:46:32 2025 +0800
[SPARK-50945][ML][PYTHON][CONNECT] Support Summarizer and SummaryBuilder on Connect
### What changes were proposed in this pull request?
Support Summarizer and SummaryBuilder on Connect
### Why are the changes needed?
For feature parity
### Does this PR introduce _any_ user-facing change?
yes, new feature supported
```
In [2]: data = [
...: [Vectors.dense([1, 0, 0, -2]), 1.0],
...: [Vectors.dense([4, 5, 0, 3]), 2.0],
...: [Vectors.dense([6, 7, 0, 8]), 1.0],
...: [Vectors.dense([9, 0, 0, 1]), 1.0],
...: ]
...: df = spark.createDataFrame(data, ["features", "weight"])
...:
...: summarizer = Summarizer.metrics("mean", "count")
In [3]: df.select(summarizer.summary(df.features)).show(truncate=False)
+--------------------------------+
|aggregate_metrics(features, 1.0)|
+--------------------------------+
|{[5.0,3.0,0.0,2.5], 4} |
+--------------------------------+
```
### How was this patch tested?
new tests
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #49847 from zhengruifeng/ml_connect_summarizer.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
(cherry picked from commit 553149732aef68da28208eb81e163b066485bf1c)
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
.../org/apache/spark/ml/stat/Summarizer.scala | 35 +++++++++++-
.../sql/ml/InternalFunctionRegistration.scala | 4 ++
python/pyspark/ml/stat.py | 30 +++++++++-
python/pyspark/ml/tests/test_stat.py | 66 +++++++++++++++++++++-
.../sql/catalyst/analysis/FunctionRegistry.scala | 2 +-
5 files changed, 130 insertions(+), 7 deletions(-)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala b/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala
index ac8845b09069..f4718b203431 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala
@@ -26,13 +26,15 @@ import org.apache.spark.ml.linalg.{Vector, Vectors, VectorUDT}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Column
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes}
+import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes, Literal}
import org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate
import org.apache.spark.sql.catalyst.trees.BinaryLike
+import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.classic.ClassicConversions.ColumnConstructorExt
import org.apache.spark.sql.classic.ExpressionUtils.expression
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.Utils
/**
@@ -290,6 +292,14 @@ private[spark] object SummaryBuilderImpl extends Logging {
StructType(fields)
}
+ private def extractRequestedMetrics(metrics: Expression): (Seq[Metric], Seq[ComputeMetric]) = {
+ metrics.eval() match {
+ case arrayData: ArrayData =>
+ val requested = arrayData.toSeq[UTF8String](StringType)
+ getRelevantMetrics(requested.map(_.toString))
+ }
+ }
+
private val vectorUDT = new VectorUDT
/**
@@ -342,7 +352,7 @@ private[spark] object SummaryBuilderImpl extends Logging {
private[stat] case object ComputeMin extends ComputeMetric
- private case class MetricsAggregate(
+ private[spark] case class MetricsAggregate(
requestedMetrics: Seq[Metric],
requestedComputeMetrics: Seq[ComputeMetric],
featuresExpr: Expression,
@@ -353,6 +363,27 @@ private[spark] object SummaryBuilderImpl extends Logging {
with ImplicitCastInputTypes
with BinaryLike[Expression] {
+ // helper constructor
+ def this(
+ metrics: (Seq[Metric], Seq[ComputeMetric]),
+ featuresExpr: Expression,
+ weightExpr: Expression) = {
+ this(metrics._1, metrics._2, featuresExpr, weightExpr, 0, 0)
+ }
+
+ def this(
+ requestedMetrics: Expression,
+ featuresExpr: Expression,
+ weightExpr: Expression) = {
+ this(extractRequestedMetrics(requestedMetrics), featuresExpr, weightExpr)
+ }
+
+ def this(
+ requestedMetrics: Expression,
+ featuresExpr: Expression) = {
+ this(requestedMetrics, featuresExpr, Literal(1.0))
+ }
+
override def eval(state: SummarizerBuffer): Any = {
val metrics = requestedMetrics.map {
case Mean => vectorUDT.serialize(state.mean)
diff --git a/mllib/src/main/scala/org/apache/spark/sql/ml/InternalFunctionRegistration.scala b/mllib/src/main/scala/org/apache/spark/sql/ml/InternalFunctionRegistration.scala
index 2141d4251ede..173f3d4f99c1 100644
--- a/mllib/src/main/scala/org/apache/spark/sql/ml/InternalFunctionRegistration.scala
+++ b/mllib/src/main/scala/org/apache/spark/sql/ml/InternalFunctionRegistration.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.ml
import org.apache.spark.ml.linalg.{SparseVector, Vector, Vectors}
+import org.apache.spark.ml.stat._
import org.apache.spark.mllib.linalg.{SparseVector => OldSparseVector, Vector => OldVector}
import org.apache.spark.sql.{SparkSessionExtensions, SparkSessionExtensionsProvider}
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
@@ -96,6 +97,9 @@ object InternalFunctionRegistration {
case exprs =>
throw QueryCompilationErrors.wrongNumArgsError("array_to_vector", "1", exprs.size)
}
+
+ FunctionRegistry
+ .registerInternalExpression[SummaryBuilderImpl.MetricsAggregate]("aggregate_metrics")
}
class InternalFunctionRegistration extends SparkSessionExtensionsProvider {
diff --git a/python/pyspark/ml/stat.py b/python/pyspark/ml/stat.py
index b9d40d46fd09..fd32a1515101 100644
--- a/python/pyspark/ml/stat.py
+++ b/python/pyspark/ml/stat.py
@@ -416,6 +416,17 @@ class Summarizer:
@staticmethod
def _get_single_metric(col: Column, weightCol: Optional[Column], metric: str) -> Column:
col, weightCol = Summarizer._check_param(col, weightCol)
+
+ if is_remote():
+ # The alias name maybe different from the one in Spark Classic,
+ # because we cannot get the same string representation of the Column object.
+ return (
+ Summarizer.metrics(metric)
+ .summary(col, weightCol)
+ .getField(metric)
+ .alias(f"{metric}({col._expr})")
+ )
+
return Column(
JavaWrapper._new_java_obj(
"org.apache.spark.ml.stat.Summarizer." + metric, col._jc, weightCol._jc
@@ -457,6 +468,12 @@ class Summarizer:
-------
:py:class:`pyspark.ml.stat.SummaryBuilder`
"""
+ if is_remote():
+ builder = SummaryBuilder(None)
+ builder._metrics = [m for m in metrics] # type: ignore[attr-defined]
+ builder._java_obj = None
+ return builder
+
from pyspark.core.context import SparkContext
from pyspark.sql.classic.column import _to_seq
@@ -481,7 +498,8 @@ class SummaryBuilder(JavaWrapper):
"""
def __init__(self, jSummaryBuilder: "JavaObject"):
- super(SummaryBuilder, self).__init__(jSummaryBuilder)
+ if not is_remote():
+ super(SummaryBuilder, self).__init__(jSummaryBuilder)
def summary(self, featuresCol: Column, weightCol: Optional[Column] = None) -> Column:
"""
@@ -503,6 +521,16 @@ class SummaryBuilder(JavaWrapper):
an aggregate column that contains the statistics. The exact content of this
structure is determined during the creation of the builder.
"""
+ if is_remote():
+ from pyspark.sql.connect.functions import builtin as F
+
+ return F._invoke_function(
+ "aggregate_metrics",
+ F.array([F.lit(m) for m in self._metrics]), # type: ignore[attr-defined]
+ featuresCol,
+ weightCol if weightCol is not None else F.lit(1.0),
+ )
+
featuresCol, weightCol = Summarizer._check_param(featuresCol, weightCol)
assert self._java_obj is not None
diff --git a/python/pyspark/ml/tests/test_stat.py b/python/pyspark/ml/tests/test_stat.py
index bd827e3a27f7..5841fd0c99fc 100644
--- a/python/pyspark/ml/tests/test_stat.py
+++ b/python/pyspark/ml/tests/test_stat.py
@@ -18,13 +18,73 @@
import numpy as np
import unittest
-from pyspark.ml.linalg import Vectors
-from pyspark.ml.stat import ChiSquareTest, Correlation, KolmogorovSmirnovTest
-from pyspark.sql import DataFrame
+from pyspark.ml.linalg import Vectors, DenseVector
+from pyspark.ml.stat import (
+ ChiSquareTest,
+ Correlation,
+ KolmogorovSmirnovTest,
+ Summarizer,
+ SummaryBuilder,
+)
+from pyspark.sql import functions as F
+from pyspark.sql import DataFrame, Row
from pyspark.testing.sqlutils import ReusedSQLTestCase
class StatTestsMixin:
+ def test_summarizer(self):
+ spark = self.spark
+ data = [
+ [Vectors.dense([1, 0, 0, -2]), 1.0],
+ [Vectors.dense([4, 5, 0, 3]), 2.0],
+ [Vectors.dense([6, 7, 0, 8]), 1.0],
+ [Vectors.dense([9, 0, 0, 1]), 1.0],
+ ]
+ df = spark.createDataFrame(data, ["features", "weight"])
+
+ summarizer = Summarizer.metrics("mean", "count")
+ self.assertIsInstance(summarizer, SummaryBuilder)
+
+ res1 = df.select(summarizer.summary(df.features))
+ self.assertEqual(res1.columns, ["aggregate_metrics(features, 1.0)"])
+ self.assertEqual(res1.count(), 1)
+ self.assertEqual(res1.head()[0], Row(mean=DenseVector([5.0, 3.0, 0.0, 2.5]), count=4))
+
+ res2 = df.select(summarizer.summary(F.col("features"), df.weight))
+ self.assertEqual(res2.columns, ["aggregate_metrics(features, weight)"])
+ self.assertEqual(res2.count(), 1)
+ self.assertEqual(
+ res2.head()[0],
+ Row(mean=DenseVector([4.8, 3.4, 0.0, 2.6]), count=4),
+ res2.head()[0][0].toArray(),
+ )
+
+ res3 = df.select(Summarizer.max(df.features, df.weight))
+ self.assertEqual(res3.columns, ["max(features)"])
+ self.assertEqual(res3.count(), 1)
+ self.assertEqual(res3.head()[0], DenseVector([9.0, 7.0, 0.0, 8.0]))
+
+ res4 = df.select(Summarizer.numNonZeros(F.col("features")))
+ self.assertEqual(res4.columns, ["numNonZeros(features)"])
+ self.assertEqual(res4.count(), 1)
+ self.assertEqual(res4.head()[0], DenseVector([4.0, 2.0, 0.0, 4.0]))
+
+ res5 = df.select(Summarizer.normL1(F.col("features")))
+ self.assertEqual(res5.columns, ["normL1(features)"])
+ self.assertEqual(res5.count(), 1)
+ self.assertEqual(res5.head()[0], DenseVector([20.0, 12.0, 0.0, 14.0]))
+
+ res6 = df.select(Summarizer.normL2(F.col("features")))
+ self.assertEqual(res6.columns, ["normL2(features)"])
+ self.assertEqual(res6.count(), 1)
+ self.assertTrue(
+ np.allclose(
+ res6.head()[0].toArray(),
+ [11.5758369, 8.60232527, 0.0, 8.83176087],
+ atol=1e-4,
+ ),
+ )
+
def test_chisquaretest(self):
spark = self.spark
data = [
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index 54f6820d2091..d59352990b2c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -905,7 +905,7 @@ object FunctionRegistry {
/** Registry for internal functions used by Connect and the Column API. */
private[sql] val internal: SimpleFunctionRegistry = new SimpleFunctionRegistry
- private def registerInternalExpression[T <: Expression : ClassTag](
+ private[spark] def registerInternalExpression[T <: Expression : ClassTag](
name: String,
setAlias: Boolean = false): Unit = {
val (info, builder) = FunctionRegistryBase.build[T](name, None)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org