Github user staple commented on a diff in the pull request:
https://github.com/apache/spark/pull/2491#discussion_r17866654
--- Diff:
mllib/src/main/scala/org/apache/spark/mllib/classification/NaiveBayes.scala ---
@@ -141,7 +214,41 @@ class NaiveBayes private (private var lambda: Double)
extends Serializable with
i += 1
}
- new NaiveBayesModel(labels, pi, theta)
+ new LocalNaiveBayesModel(labels, pi, theta)
+ }
+
+ private def trainDistModel(labelAggregates: RDD[(Double, (Long,
BDV[Double]))]) = {
+ case class LabelAggregate(label: Double, numDocuments: Long,
sumFeatures: BDV[Double])
+ val aggregated = labelAggregates.map(x => LabelAggregate(x._1,
x._2._1, x._2._2))
+
+ // Compute the model's prior (pi) vector and conditional (theta)
matrix for each batch of
+ // labels.
+ // NOTE In contrast to the local trainer, the piLogDenom normalization
term is omitted here.
+ // Computing this term requires an additional aggregation on
'aggregated', and because the
+ // term is an additive constant it does not affect maximum a
posteriori model prediction.
+ val modelBlocks = aggregated.mapPartitions(p => p.grouped(100).map {
batch =>
--- End diff --
The batch size of 100 here is just an arbitrary value right now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
