This is an automated email from the ASF dual-hosted git repository.
zhengruifeng pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new e1a4f6077104 [SPARK-58964][ML] Consolidate vector and matrix UDT
column conversion
e1a4f6077104 is described below
commit e1a4f607710468e77670e118c05f25956fa2af98
Author: Ruifeng Zheng <[email protected]>
AuthorDate: Wed Aug 26 11:27:26 2026 +0800
[SPARK-58964][ML] Consolidate vector and matrix UDT column conversion
### What changes were proposed in this pull request?
This PR consolidates UDT column selection in these `MLUtils` conversion
helpers:
- `MLUtils.convertVectorColumnsToML`
- `MLUtils.convertVectorColumnsFromML`
- `MLUtils.convertMatrixColumnsToML`
- `MLUtils.convertMatrixColumnsFromML`
When columns are not explicitly specified, each helper preserves the
previous automatic selection
behavior: `ToML` helpers select only old `mllib` UDT columns, and `FromML`
helpers select only new
`ml` UDT columns. For explicit columns, the helpers validate that the named
columns exist and rely
on the existing `wrap_udt(unwrap_udt(...))` expression analysis to reject
non-UDT inputs.
Converted columns are replaced with the metadata-preserving `withColumns`
overload instead of
rebuilding the whole projection manually.
### Why are the changes needed?
After vector and matrix conversion moved to `wrap_udt` / `unwrap_udt`, the
explicit-column paths no
longer need separate source-type and target-type branches when deciding
whether a column can be
converted.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Ran:
```
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt 'mllib/testOnly
org.apache.spark.mllib.util.MLUtilsSuite -- -z "convertVectorColumnsToML" -z
"convertVectorColumnsFromML" -z "convert nullable vector columns" -z
"convertMatrixColumnsToML" -z "convertMatrixColumnsFromML" -z "convert nullable
matrix columns"'
```
Also ran:
- `git diff --check`
- line-length scan for the changed Scala files
- non-ASCII scan for the changed Scala files
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
Closes #58241 from zhengruifeng/ml-consolidate-matrix-udt-conversion-dev-3.
Authored-by: Ruifeng Zheng <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>
(cherry picked from commit e61acc6a8f4b7d15940ba965e09251f7b37e0a4a)
Signed-off-by: Ruifeng Zheng <[email protected]>
---
.../org/apache/spark/mllib/util/MLUtils.scala | 110 +++++++++------------
1 file changed, 45 insertions(+), 65 deletions(-)
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/util/MLUtils.scala
b/mllib/src/main/scala/org/apache/spark/mllib/util/MLUtils.scala
index bbfb529bdae6..31ddf5b5a36b 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/util/MLUtils.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/util/MLUtils.scala
@@ -36,6 +36,7 @@ import
org.apache.spark.sql.execution.datasources.text.TextFileFormat
import org.apache.spark.sql.functions.{col, length, lit, not, printf,
raise_error, trim,
unwrap_udt, when, wrap_udt}
import org.apache.spark.storage.StorageLevel
+import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.random.BernoulliCellSampler
/**
@@ -320,15 +321,15 @@ object MLUtils extends Logging {
* Converts vector columns in an input Dataset from the
[[org.apache.spark.mllib.linalg.Vector]]
* type to the new [[org.apache.spark.ml.linalg.Vector]] type under the
`spark.ml` package.
* @param dataset input dataset
- * @param cols a list of vector columns to be converted. New vector columns
will be ignored. If
- * unspecified, all old vector columns will be converted except
nested ones.
+ * @param cols a list of vector columns to be converted. If unspecified, all
old vector columns
+ * will be converted except nested ones.
* @return the input `DataFrame` with old vector columns converted to the
new vector type
*/
@Since("2.0.0")
@varargs
def convertVectorColumnsToML(dataset: Dataset[_], cols: String*): DataFrame
= {
val schema = dataset.schema
- val colSet = if (cols.nonEmpty) {
+ val colNames = if (cols.nonEmpty) {
cols.flatMap { c =>
val dataType = schema(c).dataType
if (dataType.getClass == classOf[VectorUDT]) {
@@ -339,43 +340,38 @@ object MLUtils extends Logging {
s"Column $c must be old Vector type to be converted to new type
but got $dataType.")
None
}
- }.toSet
+ }
} else {
schema.fields
.filter(_.dataType.getClass == classOf[VectorUDT])
.map(_.name)
- .toSet
+ .toImmutableArraySeq
}
- if (colSet.isEmpty) {
+ if (colNames.isEmpty) {
return dataset.toDF()
}
- val exprs = schema.fields.map { field =>
- val c = field.name
- if (colSet.contains(c)) {
- wrap_udt(unwrap_udt(col(c)), new MLVectorUDT).as(c, field.metadata)
- } else {
- col(c)
- }
- }
- import org.apache.spark.util.ArrayImplicits._
- dataset.select(exprs.toImmutableArraySeq: _*)
+ val fields = colNames.map(schema(_))
+ dataset.withColumns(
+ fields.map(_.name),
+ fields.map(field => wrap_udt(unwrap_udt(col(field.name)), new
MLVectorUDT)),
+ fields.map(_.metadata))
}
/**
* Converts vector columns in an input Dataset to the
[[org.apache.spark.mllib.linalg.Vector]]
* type from the new [[org.apache.spark.ml.linalg.Vector]] type under the
`spark.ml` package.
* @param dataset input dataset
- * @param cols a list of vector columns to be converted. Old vector columns
will be ignored. If
- * unspecified, all new vector columns will be converted except
nested ones.
+ * @param cols a list of vector columns to be converted. If unspecified, all
new vector columns
+ * will be converted except nested ones.
* @return the input `DataFrame` with new vector columns converted to the
old vector type
*/
@Since("2.0.0")
@varargs
def convertVectorColumnsFromML(dataset: Dataset[_], cols: String*):
DataFrame = {
val schema = dataset.schema
- val colSet = if (cols.nonEmpty) {
+ val colNames = if (cols.nonEmpty) {
cols.flatMap { c =>
val dataType = schema(c).dataType
if (dataType.getClass == classOf[MLVectorUDT]) {
@@ -386,43 +382,38 @@ object MLUtils extends Logging {
s"Column $c must be new Vector type to be converted to old type
but got $dataType.")
None
}
- }.toSet
+ }
} else {
schema.fields
.filter(_.dataType.getClass == classOf[MLVectorUDT])
.map(_.name)
- .toSet
+ .toImmutableArraySeq
}
- if (colSet.isEmpty) {
+ if (colNames.isEmpty) {
return dataset.toDF()
}
- val exprs = schema.fields.map { field =>
- val c = field.name
- if (colSet.contains(c)) {
- wrap_udt(unwrap_udt(col(c)), new VectorUDT).as(c, field.metadata)
- } else {
- col(c)
- }
- }
- import org.apache.spark.util.ArrayImplicits._
- dataset.select(exprs.toImmutableArraySeq: _*)
+ val fields = colNames.map(schema(_))
+ dataset.withColumns(
+ fields.map(_.name),
+ fields.map(field => wrap_udt(unwrap_udt(col(field.name)), new
VectorUDT)),
+ fields.map(_.metadata))
}
/**
* Converts Matrix columns in an input Dataset from the
[[org.apache.spark.mllib.linalg.Matrix]]
* type to the new [[org.apache.spark.ml.linalg.Matrix]] type under the
`spark.ml` package.
* @param dataset input dataset
- * @param cols a list of matrix columns to be converted. New matrix columns
will be ignored. If
- * unspecified, all old matrix columns will be converted except
nested ones.
+ * @param cols a list of matrix columns to be converted. If unspecified, all
old matrix columns
+ * will be converted except nested ones.
* @return the input `DataFrame` with old matrix columns converted to the
new matrix type
*/
@Since("2.0.0")
@varargs
def convertMatrixColumnsToML(dataset: Dataset[_], cols: String*): DataFrame
= {
val schema = dataset.schema
- val colSet = if (cols.nonEmpty) {
+ val colNames = if (cols.nonEmpty) {
cols.flatMap { c =>
val dataType = schema(c).dataType
if (dataType.getClass == classOf[MatrixUDT]) {
@@ -433,43 +424,38 @@ object MLUtils extends Logging {
s"Column $c must be old Matrix type to be converted to new type
but got $dataType.")
None
}
- }.toSet
+ }
} else {
schema.fields
.filter(_.dataType.getClass == classOf[MatrixUDT])
.map(_.name)
- .toSet
+ .toImmutableArraySeq
}
- if (colSet.isEmpty) {
+ if (colNames.isEmpty) {
return dataset.toDF()
}
- val exprs = schema.fields.map { field =>
- val c = field.name
- if (colSet.contains(c)) {
- wrap_udt(unwrap_udt(col(c)), new MLMatrixUDT).as(c, field.metadata)
- } else {
- col(c)
- }
- }
- import org.apache.spark.util.ArrayImplicits._
- dataset.select(exprs.toImmutableArraySeq: _*)
+ val fields = colNames.map(schema(_))
+ dataset.withColumns(
+ fields.map(_.name),
+ fields.map(field => wrap_udt(unwrap_udt(col(field.name)), new
MLMatrixUDT)),
+ fields.map(_.metadata))
}
/**
* Converts matrix columns in an input Dataset to the
[[org.apache.spark.mllib.linalg.Matrix]]
* type from the new [[org.apache.spark.ml.linalg.Matrix]] type under the
`spark.ml` package.
* @param dataset input dataset
- * @param cols a list of matrix columns to be converted. Old matrix columns
will be ignored. If
- * unspecified, all new matrix columns will be converted except
nested ones.
+ * @param cols a list of matrix columns to be converted. If unspecified, all
new matrix columns
+ * will be converted except nested ones.
* @return the input `DataFrame` with new matrix columns converted to the
old matrix type
*/
@Since("2.0.0")
@varargs
def convertMatrixColumnsFromML(dataset: Dataset[_], cols: String*):
DataFrame = {
val schema = dataset.schema
- val colSet = if (cols.nonEmpty) {
+ val colNames = if (cols.nonEmpty) {
cols.flatMap { c =>
val dataType = schema(c).dataType
if (dataType.getClass == classOf[MLMatrixUDT]) {
@@ -480,31 +466,25 @@ object MLUtils extends Logging {
s"Column $c must be new Matrix type to be converted to old type
but got $dataType.")
None
}
- }.toSet
+ }
} else {
schema.fields
.filter(_.dataType.getClass == classOf[MLMatrixUDT])
.map(_.name)
- .toSet
+ .toImmutableArraySeq
}
- if (colSet.isEmpty) {
+ if (colNames.isEmpty) {
return dataset.toDF()
}
- val exprs = schema.fields.map { field =>
- val c = field.name
- if (colSet.contains(c)) {
- wrap_udt(unwrap_udt(col(c)), new MatrixUDT).as(c, field.metadata)
- } else {
- col(c)
- }
- }
- import org.apache.spark.util.ArrayImplicits._
- dataset.select(exprs.toImmutableArraySeq: _*)
+ val fields = colNames.map(schema(_))
+ dataset.withColumns(
+ fields.map(_.name),
+ fields.map(field => wrap_udt(unwrap_udt(col(field.name)), new
MatrixUDT)),
+ fields.map(_.metadata))
}
-
/**
* Returns the squared Euclidean distance between two vectors. The following
formula will be used
* if it does not introduce too much numerical error:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
