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/01/22 12:09:28 UTC
(spark) branch branch-4.0 updated: [SPARK-50948][ML][PYTHON][CONNECT] Add support for StringIndexer/PCA 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 2631fe09a162 [SPARK-50948][ML][PYTHON][CONNECT] Add support for StringIndexer/PCA on Connect
2631fe09a162 is described below
commit 2631fe09a1627ac2f17dd19ebe7342afecbca2e3
Author: Bobby Wang <wb...@gmail.com>
AuthorDate: Wed Jan 22 20:08:55 2025 +0800
[SPARK-50948][ML][PYTHON][CONNECT] Add support for StringIndexer/PCA on Connect
### What changes were proposed in this pull request?
This PR adds support for PCA and StringIndexer on Connect:
### Why are the changes needed?
for feature parity
### Does this PR introduce _any_ user-facing change?
yes, new algorithms supported on connect
### How was this patch tested?
The newly added tests pass
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #49600 from wbo4958/string_indexer_pca.
Authored-by: Bobby Wang <wb...@gmail.com>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
(cherry picked from commit 66588461948192dc0763118627db1fceb7e3e180)
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
.../services/org.apache.spark.ml.Estimator | 2 +
.../services/org.apache.spark.ml.Transformer | 2 +
.../scala/org/apache/spark/ml/feature/PCA.scala | 5 +
.../apache/spark/ml/feature/StringIndexer.scala | 4 +
python/pyspark/ml/tests/test_feature.py | 109 +++++++++++++++++++++
5 files changed, 122 insertions(+)
diff --git a/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Estimator b/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Estimator
index 6221c61a0670..2cc3c3b6382a 100644
--- a/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Estimator
+++ b/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Estimator
@@ -50,3 +50,5 @@ org.apache.spark.ml.feature.StandardScaler
org.apache.spark.ml.feature.MaxAbsScaler
org.apache.spark.ml.feature.MinMaxScaler
org.apache.spark.ml.feature.RobustScaler
+org.apache.spark.ml.feature.StringIndexer
+org.apache.spark.ml.feature.PCA
diff --git a/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Transformer b/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Transformer
index 47e9f4b79e34..886c4cca209b 100644
--- a/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Transformer
+++ b/mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Transformer
@@ -48,3 +48,5 @@ org.apache.spark.ml.feature.StandardScalerModel
org.apache.spark.ml.feature.MaxAbsScalerModel
org.apache.spark.ml.feature.MinMaxScalerModel
org.apache.spark.ml.feature.RobustScalerModel
+org.apache.spark.ml.feature.StringIndexerModel
+org.apache.spark.ml.feature.PCAModel
diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/PCA.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/PCA.scala
index 0bd9a3c38a1e..7d630233eb0a 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/feature/PCA.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/feature/PCA.scala
@@ -127,6 +127,11 @@ class PCAModel private[ml] (
import PCAModel._
+ // For ml connect only
+ @Since("4.0.0")
+ private[ml] def this() = this(Identifiable.randomUID("pca"),
+ DenseMatrix.zeros(1, 1), Vectors.empty.asInstanceOf[DenseVector])
+
/** @group setParam */
@Since("1.5.0")
def setInputCol(value: String): this.type = set(inputCol, value)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
index 1acffa471e9a..8b643372ec93 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/feature/StringIndexer.scala
@@ -301,6 +301,10 @@ class StringIndexerModel (
@Since("3.0.0")
def this(labelsArray: Array[Array[String]]) = this(Identifiable.randomUID("strIdx"), labelsArray)
+ // For ml connect only
+ @Since("4.0.0")
+ private[ml] def this() = this(labels = Array.empty)
+
@deprecated("`labels` is deprecated and will be removed in 3.1.0. Use `labelsArray` " +
"instead.", "3.0.0")
@Since("1.5.0")
diff --git a/python/pyspark/ml/tests/test_feature.py b/python/pyspark/ml/tests/test_feature.py
index dba4afc213ba..5e69e60be908 100644
--- a/python/pyspark/ml/tests/test_feature.py
+++ b/python/pyspark/ml/tests/test_feature.py
@@ -18,6 +18,7 @@
import tempfile
import unittest
+from typing import List, Tuple, Any
import numpy as np
@@ -43,6 +44,8 @@ from pyspark.ml.feature import (
TargetEncoder,
VectorSizeHint,
VectorAssembler,
+ PCA,
+ PCAModel,
)
from pyspark.ml.linalg import DenseVector, SparseVector, Vectors
from pyspark.sql import Row
@@ -51,6 +54,112 @@ from pyspark.testing.mlutils import check_params, SparkSessionTestCase
class FeatureTestsMixin:
+ def test_string_indexer(self):
+ df = (
+ self.spark.createDataFrame(
+ [
+ (1, "a", "e"),
+ (2, "b", "f"),
+ (3, "c", "e"),
+ (4, "a", "f"),
+ (5, "a", "f"),
+ (6, "c", "f"),
+ ],
+ ["id", "label1", "label2"],
+ )
+ .coalesce(1)
+ .sortWithinPartitions("id")
+ )
+ # single input
+ si = StringIndexer(inputCol="label1", outputCol="index1")
+ model = si.fit(df.select("label1"))
+
+ # read/write
+ with tempfile.TemporaryDirectory(prefix="read_write") as tmp_dir:
+ si.write().overwrite().save(tmp_dir)
+ si2 = StringIndexer.load(tmp_dir)
+ self.assertEqual(str(si), str(si2))
+ self.assertEqual(si.getInputCol(), "label1")
+ self.assertEqual(si2.getInputCol(), "label1")
+
+ model.write().overwrite().save(tmp_dir)
+ model2 = StringIndexerModel.load(tmp_dir)
+ self.assertEqual(str(model), str(model2))
+ self.assertEqual(model.getInputCol(), "label1")
+ self.assertEqual(model.getOutputCol(), "index1")
+ self.assertEqual(model2.getInputCol(), "label1")
+
+ indexed_df = model.transform(df.select("label1"))
+ self.assertEqual(sorted(indexed_df.schema.names), sorted(["label1", "index1"]))
+
+ def check_a_b(result: List[Tuple[Any, Any]]) -> None:
+ self.assertTrue(result[0][0] == "a" and result[1][0] == "b" and result[2][0] == "c")
+ sorted_value = sorted([v for _, v in result])
+ self.assertEqual(sorted_value, [0.0, 1.0, 2.0])
+
+ check_a_b(sorted(set([(i[0], i[1]) for i in indexed_df.collect()]), key=lambda x: x[0]))
+
+ # multiple inputs
+ input_cols = ["label1", "label2"]
+ output_cols = ["index1", "index2"]
+ si = StringIndexer(inputCols=input_cols, outputCols=output_cols)
+ model = si.fit(df.select(*input_cols))
+ self.assertEqual(model.getInputCols(), input_cols)
+ self.assertEqual(model.getOutputCols(), output_cols)
+
+ indexed_df = model.transform(df.select(*input_cols))
+ self.assertEqual(
+ sorted(indexed_df.schema.names), sorted(["label1", "index1", "label2", "index2"])
+ )
+
+ rows = indexed_df.collect()
+ check_a_b(sorted(set([(i[0], i[2]) for i in rows]), key=lambda x: x[0]))
+
+ # check e f
+ result = sorted(set([(i[1], i[3]) for i in rows]), key=lambda x: x[0])
+ self.assertTrue(result[0][0] == "e" and result[1][0] == "f")
+ sorted_value = sorted([v for _, v in result])
+ self.assertEqual(sorted_value, [0.0, 1.0])
+
+ def test_pca(self):
+ df = self.spark.createDataFrame(
+ [
+ (Vectors.sparse(5, [(1, 1.0), (3, 7.0)]),),
+ (Vectors.dense([2.0, 0.0, 3.0, 4.0, 5.0]),),
+ (Vectors.dense([4.0, 0.0, 0.0, 6.0, 7.0]),),
+ ],
+ ["features"],
+ )
+ pca = PCA(k=2, inputCol="features", outputCol="pca_features")
+
+ model = pca.fit(df)
+ self.assertEqual(model.getK(), 2)
+ self.assertTrue(
+ np.allclose(model.explainedVariance.toArray(), [0.79439, 0.20560], atol=1e-4)
+ )
+ model.setOutputCol("output")
+ # Transform the data using the PCA model
+ transformed_df = model.transform(df)
+ self.assertTrue(
+ np.allclose(
+ transformed_df.collect()[0].output.toArray(), [1.64857, -4.013282], atol=1e-4
+ )
+ )
+
+ # read/write
+ with tempfile.TemporaryDirectory(prefix="read_write") as tmp_dir:
+ pca.write().overwrite().save(tmp_dir)
+ pca2 = PCA.load(tmp_dir)
+ self.assertEqual(str(pca), str(pca2))
+ self.assertEqual(pca.getInputCol(), "features")
+ self.assertEqual(pca2.getInputCol(), "features")
+
+ model.write().overwrite().save(tmp_dir)
+ model2 = PCAModel.load(tmp_dir)
+ self.assertEqual(str(model), str(model2))
+ self.assertEqual(model.getInputCol(), "features")
+ self.assertEqual(model2.getInputCol(), "features")
+
def test_vector_assembler(self):
# Create a DataFrame
df = (
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org