java - 将 JavaRDD 转换为 JavaRDD

Question

我正在尝试对维基百科 XML 转储执行 LDA。在获得原始文本的 RDD 之后，我创建了一个数据框并通过 Tokenizer、StopWords 和 CountVectorizer 管道对其进行转换。我打算将 Vectors 输出的 RDD 从 CountVectorizer 传递到 MLLib 中的 OnlineLDA。这是我的代码:

 // Configure an ML pipeline
 RegexTokenizer tokenizer = new RegexTokenizer()
   .setInputCol("text")
   .setOutputCol("words");

 StopWordsRemover remover = new StopWordsRemover()
          .setInputCol("words")
          .setOutputCol("filtered");

 CountVectorizer cv = new CountVectorizer()
          .setVocabSize(vocabSize)
          .setInputCol("filtered")
          .setOutputCol("features");

 Pipeline pipeline = new Pipeline()
          .setStages(new PipelineStage[] {tokenizer, remover, cv});

// Fit the pipeline to train documents.
 PipelineModel model = pipeline.fit(fileDF);

 JavaRDD<Vector> countVectors = model.transform(fileDF)
          .select("features").toJavaRDD()
          .map(new Function<Row, Vector>() {
            public Vector call(Row row) throws Exception {
                Object[] arr = row.getList(0).toArray();

                double[] features = new double[arr.length];
                int i = 0;
                for(Object obj : arr){
                    features[i++] = (double)obj;
                }
                return Vectors.dense(features);
            }
          });

由于该行，我得到了类转换异常

Object[] arr = row.getList(0).toArray();


Caused by: java.lang.ClassCastException: org.apache.spark.mllib.linalg.SparseVector cannot be cast to scala.collection.Seq
at org.apache.spark.sql.Row$class.getSeq(Row.scala:278)
at org.apache.spark.sql.catalyst.expressions.GenericRow.getSeq(rows.scala:192)
at org.apache.spark.sql.Row$class.getList(Row.scala:286)
at org.apache.spark.sql.catalyst.expressions.GenericRow.getList(rows.scala:192)
at xmlProcess.ParseXML$2.call(ParseXML.java:142)
at xmlProcess.ParseXML$2.call(ParseXML.java:1)

我找到了执行此操作的 Scala 语法 here但找不到任何用 Java 来做的例子。我试过 row.getAs[Vector](0) 但这只是 Scala 语法。有什么方法可以用 Java 实现吗？

Answer 1

因此，我能够通过简单转换为 Vector 来完成此操作。我不知道为什么我没有先尝试简单的事情!

         JavaRDD<Vector> countVectors = model.transform(fileDF)
              .select("features").toJavaRDD()
              .map(new Function<Row, Vector>() {
                public Vector call(Row row) throws Exception {
                    return (Vector)row.get(0);
                }
              });