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 2026/08/10 04:41:31 UTC
(spark) branch branch-4.x updated: [SPARK-58637][ML] Optimize HashingTF transform closure
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 ce4103ca02ee [SPARK-58637][ML] Optimize HashingTF transform closure
ce4103ca02ee is described below
commit ce4103ca02ee9dccc20791fbba766e998160e141
Author: Ruifeng Zheng <ru...@apache.org>
AuthorDate: Mon Aug 10 12:41:05 2026 +0800
[SPARK-58637][ML] Optimize HashingTF transform closure
### What changes were proposed in this pull request?
This PR optimizes `HashingTF.transform` by selecting the binary/counting UDF and hash function
from `($(binary), hashFuncVersion)` before executing the UDF. The UDF calculates feature indexes
from the captured hash function and local `numFeatures` value instead of calling the instance
`indexOf` method.
Binary mode uses `OpenHashMap.update`, while counting mode uses `changeValue`. Both store integer
term counts and convert them to `Double` only when constructing the output sparse vector.
The compatibility test for HashingTF models saved before Spark 3.0 now also verifies `transform`,
including the saved binary mode and legacy hash function.
### Why are the changes needed?
Calling `indexOf` from the UDF captures the `HashingTF` transformer and performs a parameter lookup
and hash-version match for every term. Selecting the behavior before constructing the UDF reduces
the closure to the values it needs, removes repeated branching and parameter lookups, and reduces
the term-count map value size from `Double` to `Int`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`build/sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 'mllib/testOnly org.apache.spark.ml.feature.HashingTFSuite'`
All 6 tests passed. The compiled UDF helper signatures were also inspected to confirm that they do
not retain a `HashingTF` receiver.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
Closes #57845 from zhengruifeng/SPARK-58637.
Authored-by: Ruifeng Zheng <ru...@apache.org>
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
(cherry picked from commit 8210537efef425d35f4ebf58341fdc44a9ef0e25)
Signed-off-by: Ruifeng Zheng <ru...@foxmail.com>
---
.../org/apache/spark/ml/feature/HashingTF.scala | 35 ++++++++++++++++++----
.../apache/spark/ml/feature/HashingTFSuite.scala | 12 ++++++++
2 files changed, 42 insertions(+), 5 deletions(-)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/HashingTF.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/HashingTF.scala
index dab0a6494fdb..0610465fe7fa 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/feature/HashingTF.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/feature/HashingTF.scala
@@ -91,12 +91,37 @@ class HashingTF @Since("3.0.0") private[ml] (
override def transform(dataset: Dataset[_]): DataFrame = {
val outputSchema = transformSchema(dataset.schema)
val n = $(numFeatures)
- val updateFunc = if ($(binary)) (v: Double) => 1.0 else (v: Double) => v + 1.0
+ def binaryHashUDF(hashFunc: Any => Int) = udf { terms: Seq[_] =>
+ val map = new OpenHashMap[Int, Int]
+ terms.foreach { term =>
+ map.update(Utils.nonNegativeMod(hashFunc(term), n), 1)
+ }
+ Vectors.sparse(n, map.iterator.map { case (index, count) =>
+ (index, count.toDouble)
+ }.toSeq)
+ }
+
+ def countHashUDF(hashFunc: Any => Int) = udf { terms: Seq[_] =>
+ val map = new OpenHashMap[Int, Int]
+ terms.foreach { term =>
+ val index = Utils.nonNegativeMod(hashFunc(term), n)
+ map.changeValue(index, 1, _ + 1)
+ }
+ Vectors.sparse(n, map.iterator.map { case (index, count) =>
+ (index, count.toDouble)
+ }.toSeq)
+ }
- val hashUDF = udf { terms: Seq[_] =>
- val map = new OpenHashMap[Int, Double]()
- terms.foreach { term => map.changeValue(indexOf(term), 1.0, updateFunc) }
- Vectors.sparse(n, map.toSeq)
+ val hashUDF = ($(binary), hashFuncVersion) match {
+ case (true, HashingTF.SPARK_2_MURMUR3_HASH) =>
+ binaryHashUDF(OldHashingTF.murmur3Hash)
+ case (false, HashingTF.SPARK_2_MURMUR3_HASH) =>
+ countHashUDF(OldHashingTF.murmur3Hash)
+ case (true, HashingTF.SPARK_3_MURMUR3_HASH) =>
+ binaryHashUDF(FeatureHasher.murmur3Hash)
+ case (false, HashingTF.SPARK_3_MURMUR3_HASH) =>
+ countHashUDF(FeatureHasher.murmur3Hash)
+ case _ => throw new IllegalArgumentException("Illegal hash function version setting.")
}
dataset.withColumn($(outputCol), hashUDF(col($(inputCol))),
diff --git a/mllib/src/test/scala/org/apache/spark/ml/feature/HashingTFSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/feature/HashingTFSuite.scala
index 861bf1e0b129..4eac3557cb93 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/feature/HashingTFSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/feature/HashingTFSuite.scala
@@ -97,6 +97,18 @@ class HashingTFSuite extends MLTest with DefaultReadWriteTest {
assert(loadedHashingTF.indexOf("c") === mLlibHashingTF.indexOf("c"))
assert(loadedHashingTF.indexOf("d") === mLlibHashingTF.indexOf("d"))
+ mLlibHashingTF.setBinary(loadedHashingTF.getBinary)
+ val terms = "a a b b c d".split(" ").toSeq
+ val df = Seq(Tuple1(terms)).toDF("words")
+ val features = loadedHashingTF
+ .setInputCol("words")
+ .setOutputCol("features")
+ .transform(df)
+ .select("features")
+ .first()
+ .getAs[Vector](0)
+ assert(features ~== mLlibHashingTF.transform(terms).asML absTol 1e-14)
+
val metadata = spark.read.json(s"$hashingTFPath/metadata")
val sparkVersionStr = metadata.select("sparkVersion").first().getString(0)
assert(sparkVersionStr === "2.4.4")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org