fix for setOptimizer

Author: hhbyyh

mllib/src/main/scala/org/apache/spark/mllib/clustering/LDA.scala

@@ -210,14 +210,15 @@ class LDA private (
 
   /**
    * Set the LDAOptimizer used to perform the actual calculation by algorithm name.
-   * Currently "em" is supported.
+   * Currently "em", "online" is supported.
    */
   def setOptimizer(optimizerName: String): this.type = {
     this.ldaOptimizer =
       optimizerName.toLowerCase match {
         case "em" => new EMLDAOptimizer
+        case "online" => new OnlineLDAOptimizer
         case other =>
-          throw new IllegalArgumentException(s"Only em is supported but got $other.")
+          throw new IllegalArgumentException(s"Only em, online are supported but got $other.")
       }
     this
   }

mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala

@@ -396,7 +396,7 @@ class OnlineLDAOptimizer extends LDAOptimizer {
     val batchResult = statsSum :* expElogbeta
 
     // Note that this is an optimization to avoid batch.count
-    update(batchResult, iteration, (miniBatchFraction * corpusSize).toInt)
+    update(batchResult, iteration, (miniBatchFraction * corpusSize).ceil.toInt)
     this
   }