examples/mllib/JavaLatentDirichletAllocationExample.java

0001 /*
0002  * Licensed to the Apache Software Foundation (ASF) under one or more
0003  * contributor license agreements.  See the NOTICE file distributed with
0004  * this work for additional information regarding copyright ownership.
0005  * The ASF licenses this file to You under the Apache License, Version 2.0
0006  * (the "License"); you may not use this file except in compliance with
0007  * the License.  You may obtain a copy of the License at
0008  *
0009  *    http://www.apache.org/licenses/LICENSE-2.0
0010  *
0011  * Unless required by applicable law or agreed to in writing, software
0012  * distributed under the License is distributed on an "AS IS" BASIS,
0013  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014  * See the License for the specific language governing permissions and
0015  * limitations under the License.
0016  */
0017
0018 package org.apache.spark.examples.mllib;
0019
0020 import org.apache.spark.SparkConf;
0021 import org.apache.spark.api.java.JavaSparkContext;
0022
0023 // $example on$
0024 import scala.Tuple2;
0025
0026 import org.apache.spark.api.java.JavaPairRDD;
0027 import org.apache.spark.api.java.JavaRDD;
0028 import org.apache.spark.mllib.clustering.DistributedLDAModel;
0029 import org.apache.spark.mllib.clustering.LDA;
0030 import org.apache.spark.mllib.clustering.LDAModel;
0031 import org.apache.spark.mllib.linalg.Matrix;
0032 import org.apache.spark.mllib.linalg.Vector;
0033 import org.apache.spark.mllib.linalg.Vectors;
0034 // $example off$
0035
0036 public class JavaLatentDirichletAllocationExample {
0037   public static void main(String[] args) {
0038
0039     SparkConf conf = new SparkConf().setAppName("JavaKLatentDirichletAllocationExample");
0040     JavaSparkContext jsc = new JavaSparkContext(conf);
0041
0042     // $example on$
0043     // Load and parse the data
0044     String path = "data/mllib/sample_lda_data.txt";
0045     JavaRDD<String> data = jsc.textFile(path);
0046     JavaRDD<Vector> parsedData = data.map(s -> {
0047       String[] sarray = s.trim().split(" ");
0048       double[] values = new double[sarray.length];
0049       for (int i = 0; i < sarray.length; i++) {
0050         values[i] = Double.parseDouble(sarray[i]);
0051       }
0052       return Vectors.dense(values);
0053     });
0054     // Index documents with unique IDs
0055     JavaPairRDD<Long, Vector> corpus =
0056       JavaPairRDD.fromJavaRDD(parsedData.zipWithIndex().map(Tuple2::swap));
0057     corpus.cache();
0058
0059     // Cluster the documents into three topics using LDA
0060     LDAModel ldaModel = new LDA().setK(3).run(corpus);
0061
0062     // Output topics. Each is a distribution over words (matching word count vectors)
0063     System.out.println("Learned topics (as distributions over vocab of " + ldaModel.vocabSize()
0064       + " words):");
0065     Matrix topics = ldaModel.topicsMatrix();
0066     for (int topic = 0; topic < 3; topic++) {
0067       System.out.print("Topic " + topic + ":");
0068       for (int word = 0; word < ldaModel.vocabSize(); word++) {
0069         System.out.print(" " + topics.apply(word, topic));
0070       }
0071       System.out.println();
0072     }
0073
0074     ldaModel.save(jsc.sc(),
0075       "target/org/apache/spark/JavaLatentDirichletAllocationExample/LDAModel");
0076     DistributedLDAModel sameModel = DistributedLDAModel.load(jsc.sc(),
0077       "target/org/apache/spark/JavaLatentDirichletAllocationExample/LDAModel");
0078     // $example off$
0079
0080     jsc.stop();
0081   }
0082 }