```scala  import org.apache.spark.rdd._ import org.apache.spark.mllib.clustering.{LDA, DistributedLDAModel, LocalLDAModel} import org.apache.spark.mllib.linalg.{Vector, Vectors} import scala.collection.mutable //create training document set val input = Seq("this is a document","this could be another document","these are training, not tests", "here is the final file (document)") val corpus: RDD[Array[String]] = sc.parallelize(input.map{    doc => doc.split("\\s") }) val termCounts: Array[(String, Long)] = corpus.flatMap(_.map(_ -> 1L)).reduceByKey(_ + _).collect().sortBy(-_._2) val vocabArray: Array[String] = termCounts.takeRight(termCounts.size).map(_._1) val vocab: Map[String, Int] = vocabArray.zipWithIndex.toMap // Convert training documents into term count vectors val documents: RDD[(Long, Vector)] =     corpus.zipWithIndex.map { case (tokens, id) =>         val counts = new mutable...