Spark LDA with scala and Java

Question

In current, I have try implemented LDA algorithm using Apache Spark with scala following:

    // 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()

But after that, i convert this code from scala to Java API:

        // 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();

But LDA model only accept JavaPairRDD parameter for run() function. I have stuck when try to parse countVectors as JavaPairRDD as scala code can do it. Please help me if you have another solution. Thank you very much.

EDITED: I have change my code follow:

        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();

Many thank to @Till Rohrmann. But after run program, I have exception message:

Exception in thread "main" java.lang.NoSuchMethodError: org.apache.spark.sql.Column.as(Ljava/lang/String;Lorg/apache/spark/sql/types/Metadata;)Lorg/apache/spark/sql/Column; at org.apache.spark.ml.feature.StopWordsRemover.transform(StopWordsRemover.scala:144)

Can you help me to resolve this?

Answer 1

You can use the mapToPair method to create a JavaPairRDD.

Assuming that TextId has a String and a Long field the code could look the following:

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();