用scala和Java开发LDA

Question

目前，我尝试用Apache Spark和scala实现LDA算法：

    // Filter out stopwords
val stopwords: Array[String] = sc.textFile("data/english_stops_words.txt").collect()
val filteredTokens = new StopWordsRemover()
  .setStopWords(stopwords)
  .setCaseSensitive(false)
  .setInputCol("words")
  .setOutputCol("filtered")
  .transform(tokens)

// Limit to top `vocabSize` most common words and convert to word count vector features
val cvModel = new CountVectorizer()
  .setInputCol("filtered")
  .setOutputCol("features")
  .setVocabSize(vocabSize)
  .fit(filteredTokens)
val countVectors = cvModel.transform(filteredTokens)
  .select("docId", "features")
  .map { case Row(docId: Long, countVector: Vector) => (docId, countVector) }
  .cache()

        // Filter out stopwords
    List<String> stopwords = sc.textFile("data/english_stops_words.txt")
            .collect();
    DataFrame filteredTokens = new StopWordsRemover()
            .setStopWords(stopwords.toArray(new String[0]))
            .setCaseSensitive(false).setInputCol("words")
            .setOutputCol("filtered").transform(tokens);

    // Limit to top `vocabSize` most common words and convert to word count
    // vector features
    CountVectorizerModel cvModel = new CountVectorizer()
            .setInputCol("filtered").setOutputCol("features")
            .setVocabSize(vocabSize).fit(filteredTokens);

    JavaRDD<TextId> countVectors = cvModel.transform(filteredTokens)
              .select("docId", "features").toJavaRDD()
              .map(new Function<Row, TextId>() {

                private static final long serialVersionUID = 1L;

                @Override
                public TextId call(Row row) throws Exception {

                    return new TextId(row.get(0).toString(), Long.parseLong(row.get(1).toString()));
                }
            }).cache();

        JavaPairRDD<Long, Vector> countVectors = cvModel.transform(filteredTokens)
              .select("docId", "features").toJavaRDD()
              .mapToPair(new PairFunction<Row, Long, Vector>() {
                public Tuple2<Long, Vector> call(Row row) throws Exception {
                    return new Tuple2<Long, Vector>(Long.parseLong(row.getString(0)), Vectors.dense(row.getDouble(1)));
                }
            }).cache();

Answer 1

可以使用MaptoPair方法创建JavaPairrdd。

假设textid有一个字符串和一个long字段，代码可以如下所示：

JavaPairRDD<Long, Vector> countVectors = cvModel.transform(filteredTokens)
  .select("docId", "features").toJavaRDD()
  .mapToPair(new PairFunction<Row, Long, Vector>() {
    public Tuple2<Long, Vector> call(Row row) throws Exception {
        return new Tuple2<String, Long>(row.getAs[Long](0), row.getAs[Vector](1)));
    }
}).cache();