从 DataFrame 转换为 JavaPairRDD<Long, Vector>答案

【问题标题】：Convert from DataFrame to JavaPairRDD<Long, Vector>从 DataFrame 转换为 JavaPairRDD<Long, Vector>
【发布时间】：2016-01-23 08:36:36
【问题描述】：

我正在尝试使用带有 Java API 的 apache spark 来实现 LDA 算法。方法 LDA().run() 接受参数 JavaPairRDD 文档。我已经使用 scala 创建 RDD[(Long, Vector)] 跟随：

val countVectors = cvModel.transform(filteredTokens)
    .select("docId", "features")
    .map { case Row(docId: Long, countVector: Vector) => (docId, countVector) }
    .cache()

然后输入LDA：

lda.run(countVectors)

但在 Java API 中，我使用以下代码拥有 CountVectorizerModel：

CountVectorizerModel cvModel = new CountVectorizer()
        .setInputCol("filtered").setOutputCol("features")
        .setVocabSize(vocabSize).fit(filteredTokens);

看起来像这样：

(0,(22,[0,8,9,10,14,16,18],
[1.0,1.0,1.0,1.0,1.0,1.0,1.0]))
(1,(22,[0,1,2,3,4,5,6,7,11,12,13,15,17,19,20,21],
1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0]))

如果我想从cvModel转换成JavaPairRDD countVectors怎么办？我试过这个：

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

但它不起作用。尝试时出现异常：

Vectors.dense(row.getDouble(1))

所以，如果您有任何理想的从 DataFrame cvModel 转换为 JavaPairRDD 的方法，请告诉我。

我正在使用 Spark 和 MLlib 1.5.1 和 Java8

非常感谢任何帮助。谢谢这是我尝试从 DataFrame 转换为 JavaPairRDD 时的异常日志文件

15/10/25 10:03:07 ERROR Executor: Exception in task 0.0 in stage 7.0     (TID 6)
java.lang.ClassCastException: java.lang.Long cannot be cast to      java.lang.String
at org.apache.spark.sql.Row$class.getString(Row.scala:249)
at org.apache.spark.sql.catalyst.expressions.GenericRow.getString(rows.scala:191)
at UIT_LDA_ONLINE.LDAOnline$2.call(LDAOnline.java:88)
at UIT_LDA_ONLINE.LDAOnline$2.call(LDAOnline.java:1)
at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1030)
at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1030)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:278)
at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:262)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
15/10/25 10:03:07 WARN TaskSetManager: Lost task 0.0 in stage 7.0 (TID 6, localhost): java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.String
at org.apache.spark.sql.Row$class.getString(Row.scala:249)
at org.apache.spark.sql.catalyst.expressions.GenericRow.getString(rows.scala:191)
at UIT_LDA_ONLINE.LDAOnline$2.call(LDAOnline.java:88)
at UIT_LDA_ONLINE.LDAOnline$2.call(LDAOnline.java:1)
at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1030)
at org.apache.spark.api.java.JavaPairRDD$$anonfun$pairFunToScalaFun$1.apply(JavaPairRDD.scala:1030)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:278)
at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:262)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

【问题讨论】：

Protip: 它不起作用 不是很好的问题描述:)
谢谢，我将描述异常:)
好的，所以我们要去某个地方，但我们还没有到达那里。异常通常提供相当多的信息。异常的类型是什么，究竟是哪一部分引起的等等。所有这些都是为了引导而不是填满屏幕。此外，您在编辑中缺少右括号。
@zero323 你很有耐心！ ;)
@eliasah 称其为投资 :) 有一天，这里的某个人成为您的同事的可能性非零，即使不是，回答有趣且精心设计的问题总是更有趣。

标签： java apache-spark apache-spark-mllib

【解决方案1】：

现在我们有了错误堆栈，这里是错误：

您正试图从行中获取一个字符串，而您的字段是 Long，因此您需要将 row.getString(0) 替换为 row.getLong(0) 作为初学者。

一旦您更正此问题，您将遇到来自相同类型但不同级别的其他错误，我可以通过给出的信息指出，但如果您应用以下内容，您将能够解决它们：

每个字段类型的行 getter 都是特定的，您需要使用正确的 get 方法。

如果不确定需要使用什么方法，可以在DataFrame上使用printSchema方法检查每个字段的类型，然后可以进行官方文档@987654321中描述的所有类型转换@。

【讨论】：