zhengruifeng commented on code in PR #58278:
URL: https://github.com/apache/spark/pull/58278#discussion_r3859637222
##########
mllib/src/main/scala/org/apache/spark/ml/feature/CountVectorizer.scala:
##########
@@ -294,13 +294,14 @@ class CountVectorizerModel(
def setBinary(value: Boolean): this.type = set(binary, value)
/** Dictionary created from [[vocabulary]] and its indices, broadcast once
for [[transform()]] */
- private var broadcastDict: Option[Broadcast[Map[String, Int]]] = None
+ private var broadcastDict: Option[Broadcast[OpenHashMap[String, Int]]] = None
@Since("2.0.0")
override def transform(dataset: Dataset[_]): DataFrame = {
val outputSchema = transformSchema(dataset.schema, logging = true)
if (broadcastDict.isEmpty) {
- val dict = Utils.toMapWithIndex(vocabulary)
+ // Reserve 0 for missing terms, which avoids allocating an Option for
every lookup.
+ val dict = Utils.toOpenHashMapWithIndex(vocabulary, indexOffset = 1)
broadcastDict = Some(dataset.sparkSession.sparkContext.broadcast(dict))
Review Comment:
Applied in 103f82da6cc. CountVectorizer now broadcasts only a cloned
`Array[String]` and builds the pre-sized Java HashMap lazily through an
executor-local weak cache keyed by the broadcast array identity. Concurrent
tasks share the map, and the weak key lets it be reclaimed with the broadcast
value. This also removes the core Kryo registration change because
`Array[String]` already works with `spark.kryo.registrationRequired=true`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
