K-means train process on message boards

MessageBoardsKmeansTrain.scala

val messagesTable = sc.textFile(messagesInput).map(_.split(";")).map(m => MBMessage(m(4), m(5)))

val schema =
StructType(
   StructField("title", StringType, false) ::
   StructField("body", StringType, true) :: Nil)

val messagesSchemaRDD = sqlContext.applySchema(messagesTable, schema)

messagesSchemaRDD.registerTempTable("messagesTable")


val bodies = sqlContext.sql("SELECT body from messagesTable").map(_.head.toString)

val vectors = bodies.map(Utils.featurize).cache()


val model = KMeans.train(vectors, numClusters.toInt, numIterations.toInt)

sc.makeRDD(model.clusterCenters, numClusters.toInt).saveAsObjectFile(outputModelDir)