星火：我如何工会列表<RDD>到RDD

很新的火花和Scala语言，并希望工会为以下列表中的所有RDDS（List<RDD> to RDD）：星火：我如何工会列表<RDD>到RDD

val data = for (item <- paths) yield { 
     val ad_data_path = item._1 
     val ad_data = SparkCommon.sc.textFile(ad_data_path).map { 
      line => { 
       val ad_data = new AdData(line) 
       (ad_data.ad_id, ad_data) 
      } 
     }.distinct() 
    } 
val ret = SparkCommon.sc.parallelize(data).reduce(_ ++ _)

我运行代码在IntelliJ中，虽然总是得到一个错误：

ava.lang.NullPointerException 
at org.apache.spark.rdd.RDD.<init>(RDD.scala:125) 
at org.apache.spark.rdd.UnionRDD.<init>(UnionRDD.scala:59) 
at org.apache.spark.rdd.RDD.union(RDD.scala:438) 
at org.apache.spark.rdd.RDD.$plus$plus(RDD.scala:444) 
at data.GenerateData$$anonfun$load_data$1.apply(GenerateData.scala:99) 
at data.GenerateData$$anonfun$load_data$1.apply(GenerateData.scala:99) 
at scala.collection.TraversableOnce$$anonfun$reduceLeft$1.apply(TraversableOnce.scala:177) 
at scala.collection.TraversableOnce$$anonfun$reduceLeft$1.apply(TraversableOnce.scala:172) 
at scala.collection.Iterator$class.foreach(Iterator.scala:727) 
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28) 
at scala.collection.TraversableOnce$class.reduceLeft(TraversableOnce.scala:172) 
at org.apache.spark.InterruptibleIterator.reduceLeft(InterruptibleIterator.scala:28) 
at org.apache.spark.rdd.RDD$$anonfun$18.apply(RDD.scala:847) 
at org.apache.spark.rdd.RDD$$anonfun$18.apply(RDD.scala:845) 
at org.apache.spark.SparkContext$$anonfun$26.apply(SparkContext.scala:1157) 
at org.apache.spark.SparkContext$$anonfun$26.apply(SparkContext.scala:1157) 
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62) 
at org.apache.spark.scheduler.Task.run(Task.scala:54) 
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177) 
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
at java.lang.Thread.run(Thread.java:745)

任何人有任何关于错误的想法吗？感谢提前:)

来源

2015-05-25 juffun

这可能是原因，

val listA = 1 to 10 
for(i <- listA; if i%2 == 0)yield {i}

将返回向量（2,4,6），而

for(i <- listA; if i%2 == 0)yield {val c = i}

将返回向量（（），（），（），（），（））

这就是你的情况。您正在初始化ad_data但未返回成功。

至于你的问题而言，即列表[RDD]到RDD

这里是解决方案：

val listA = sc.parallelize(1 to 10) 
val listB = sc.parallelize(10 to 1 by -1)

创造的2个RDDS

val listC = List(listA,listB)

列表
转换列表[RDD]到RDD

val listD = listC.reduce(_ union _)

希望，这个回答你的问题。

来源

2015-05-25 11:55:06 Akash

非常感谢，所述您的解决方案解决了问题。 – juffun

@juffun，可以接受答案，如果解决方案为你工作:) – Akash

当然，已经接受。 – juffun

将RDD列表转换为RDD的另一种简单方法。 SparkContext有两个重载联合方法，一种接受两个RDDS等接受RDDS列表

联盟（第一，静止）联盟（RDDS：序号[RDD [T]]））

来源

2016-09-29 09:20:20

星火：我如何工会列表<RDD>到RDD

回答

相关问题