title: Spark LDA consumes too much memory
body: <p>I'm trying to use spark mllib lda to summarize my document corpus.</p>
  
  <p>My problem setting is as bellow.</p>
  
  <ul>
  <li>about 100,000 documents </li>
  <li>about 400,000 unique words</li>
  <li>100 cluster</li>
  </ul>
  
  <p>I have 16 servers (each has 20 cores and 128GB memory).</p>
  
  <p>When I execute LDA with <code>OnlineLDAOptimizer</code>, it gives out of memory error, suggesting me to increase <code>spark.driver.maxResultSize</code> like
  <strong>Total size of serialized results of 11 tasks (1302 MB) is bigger than spark.driver.maxResultSize</strong></p>
  
  <p>I increased <code>spark.driver.maxResultSize</code> to 120GB (and also <code>spark.driver.memory</code> to 120GB) and re-run LDA but no lack.
  It still says <strong>Total size of serialized results of 11 tasks (120.1 GB) is bigger than spark.driver.maxResultSize</strong></p>
  
  <p>I tried another dataset with about 100,000 unique words and it worked.</p>
  
  <p>So, how can I estimate the memory usage when using Spark mllib LDA? I couldn't find any specification in the official documentation.</p>
  
  <p>Note I used sparse　vector for constructing docuemnt <code>RDD[(Long, Vector)]</code> passed to <code>LDA.run()</code> but don't know whether spark lda can handle sparse format correctly in internal.</p>
  
  <p>(editted) I used Scala version of LDA. Not Python version.</p>
  
  <p>This may be a related issue but no clear answer was given.
  <a href="https://stackoverflow.com/questions/35090222/spark-lda-woes-prediction-and-oom-questions">Spark LDA woes - prediction and OOM questions</a></p>
  
  <p>(edited)</p>
  
  <p>This is a snippet of my code (gist).
  <a href="https://gist.github.com/lucidfrontier45/11420721c0078c5b7415" rel="nofollow noreferrer">https://gist.github.com/lucidfrontier45/11420721c0078c5b7415</a></p>
  
  <pre class="lang-scala prettyprint-override"><code>def startJob(args: RunArgs)(implicit sc: SparkContext): Unit = {
      val src = sc.textFile(args.fname, minPartitions = args.n_partitions).map(_.split("\t"))
          .flatMap {
              // input file's format is (user_id, product_name, count)
              case Array(u, p, r, t) =&gt; Some((u.toInt, p.toInt, r.toDouble))
              case _ =&gt; None
          }.persist()
  
      // Map to convert user_id or product_name into unique sequencential id
      val userid_map = src.map(_._1).distinct().zipWithIndex().collect().toMap
      val productid_map = src.map(_._2).distinct().zipWithIndex().collect().toMap
      val inverse_userid_map = userid_map.map(_.swap)
  
      // broadcat to speedup RDD map operation
      val b_userid_map = sc.broadcast(userid_map)
      val b_productid_map = sc.broadcast(productid_map)
      val b_inverse_userid_map = sc.broadcast(inverse_userid_map)
  
      // run map
      val transformed_src = src.map { case (u, p, r) =&gt;
          (b_userid_map.value(u), b_productid_map.value(p).toInt, r)
      }
  
      println("unique items = %d".format(b_productid_map.value.size))
  
      // prepare for LDA input RDD[(LONG, Vector)]
      val documents = transformed_src.map { case (u, p, r) =&gt; (u, (p, r)) }
          .groupByKey()
          .map { t =&gt; (t._1, Vectors.sparse(b_productid_map.value.size, t._2.toSeq)) }.persist()
  
      documents.count()
      src.unpersist()
  
      // run Online Variational LDA
      val ldamodel = new LDA()
          .setK(args.k)
          .setMaxIterations(args.n_iter)
          .setOptimizer("online")
          .run(documents)
          .asInstanceOf[LocalLDAModel]
  
  
      val result = ldamodel.topicDistributions(documents)
          .map { case (i, v) =&gt;
              val u = b_inverse_userid_map.value(i)
              "%d,%s".format(u, v.toArray.mkString(","))
          }
      result.saveAsTextFile(args.out)
  }
  </code></pre>
  
  <p>Actually, I use LDA for dimensional reduction of transaction data. My data is in the format of <code>(u, p, r)</code>
  where <code>u</code> is user id, <code>p</code> is product name, <code>r</code> is the number user <code>u</code> interacted with <code>p</code>. user corresponds to document and product to word in this case. Since user id and product name are arbitrary string, I converted them to unique sequential integers before submitting to LDA.</p>
  
  <p>Thank you.</p>