Spark：如何将 RDD 的 Seq 转换为 RDD答案

【问题标题】：Spark: How to transform a Seq of RDD into a RDDSpark：如何将 RDD 的 Seq 转换为 RDD
【发布时间】：2015-02-27 15:22:30
【问题描述】：

我刚刚开始使用 Spark 和 Scala

我有一个目录，里面有多个文件我使用

成功加载了它们

sc.wholeTextFiles(directory)

现在我想更上一层楼。我实际上有一个目录，其中包含包含文件的子目录。我的目标是获得一个RDD[(String,String)]，这样我就可以继续前进，其中RDD 代表文件的名称和内容。

我尝试了以下方法：

val listOfFolders = getListOfSubDirectories(rootFolder)
val input = listOfFolders.map(directory => sc.wholeTextFiles(directory))

但我有一个Seq[RDD[(String,String)]] 如何将此Seq 转换为RDD[(String,String)]？

或者也许我做的不对，我应该尝试不同的方法？

编辑：添加代码

// HADOOP VERSION
val rootFolderHDFS = "hdfs://****/"
val hdfsURI = "hdfs://****/**/"

// returns a list of folders (currently about 800)
val listOfFoldersHDFS = ListDirectoryContents.list(hdfsURI,rootFolderHDFS)
val inputHDFS = listOfFoldersHDFS.map(directory => sc.wholeTextFiles(directory))
// RDD[(String,String)]
//    val inputHDFS2 = inputHDFS.reduceRight((rdd1,rdd2) => rdd2 ++ rdd1)
val init = sc.parallelize(Array[(String, String)]())
val inputHDFS2 = inputHDFS.foldRight(init)((rdd1,rdd2) => rdd2 ++ rdd1)

// returns org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: java.lang.StackOverflowError
println(inputHDFS2.count)

【问题讨论】：

标签： scala apache-spark seq rdd flatmap

【解决方案1】：

您可以像这样减少Seq（将RDDs 与++ 连接）：

val reduced: RDD[(String, String)] = input.reduce((left, right) => left ++ right)

更多细节为什么我们可以在这里申请reduce：

++ 是关联的 - rdda ++ (rddb ++ rddc) 或 (rdda ++ rddb) ++ rddc 无关紧要
假设Seq 是非空的（否则fold 会是更好的选择，它需要一个空的RDD[(String, String)] 作为初始累加器）。

根据Seq 的确切类型，您可能会遇到stackoverflow，因此请小心并使用更大的集合进行测试，尽管对于标准库我认为它是安全的。

【讨论】：

嗨！在你回答之前我正在使用联合，并得到一个 StackOverflow 错误......现在我正在使用 ++，但我仍然得到一个......有什么问题？ org.apache.spark.SparkException：作业因阶段失败而中止：任务序列化失败：java.lang.StackOverflowError
嗯。在这种情况下，请尝试reduceRight/foldRight。它可能会避免这种情况。（您使用哪个版本的 Scala？）
我用的是 2.10.4。关于 foldRight，您将如何编写函数？（我需要从一个元素开始，但我不知道如何创建一个emptyRDD）
如果您知道reduceRight 永远不会为空，您可以使用它。对于foldRight：val init = sc.parallelize(Array[(String, String)]())，其中sc 是SparkContext。 (spark.apache.org/docs/latest/programming-guide.html)
对于 Scala 中的长列表，reduceRight 更有可能stackoverflow，但这可能不是这里发生的事情。合并大量 RDD 时可能会发生堆栈溢出，因为 RDD 保持其完整的血统。为了缩短堆栈，需要定期 .cache() 结果并让中间 RDD 超出范围。

【解决方案2】：

您可以使用路径通配符将所有目录加载到单个 RDD 中，而不是将每个目录加载到单独的 RDD 中吗？

鉴于以下目录树...

$ tree test/spark/so
test/spark/so
├── a
│   ├── text1.txt
│   └── text2.txt
└── b
    ├── text1.txt
    └── text2.txt

使用目录的通配符创建 RDD。

scala> val rdd =  sc.wholeTextFiles("test/spark/so/*/*")
rdd: org.apache.spark.rdd.RDD[(String, String)] = test/spark/so/*/ WholeTextFileRDD[16] at wholeTextFiles at <console>:37

如您所料，计数为 4。

scala> rdd.count
res9: Long = 4

scala> rdd.collect
res10: Array[(String, String)] =
Array((test/spark/so/a/text1.txt,a1
a2
a3), (test/spark/so/a/text2.txt,a3
a4
a5), (test/spark/so/b/text1.txt,b1
b2
b3), (test/spark/so/b/text2.txt,b3
b4
b5))

【讨论】：

我发现这个解决方案非常慢。它似乎是单线程或什么的。我可以更快地加载相同的数据，为每个目录创建一个 RDD 列表，但是在合并它们时我会遇到堆栈溢出的问题。
@Brian - 如果没有看到您的实现以及您正在尝试做什么，我无法理解为什么它会变慢。发一个新帖子并参考这个可能吗？

【解决方案3】：

你应该使用 spark context 提供的union

val rdds: Seq[RDD[Int]] = (1 to 100).map(i => sc.parallelize(Seq(i)))
val rdd_union: RDD[Int] = sc.union(rdds)

【讨论】：