Spark: Как превратить Seq из РДА в РД

Question

Я только начиная с искрой & 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) 

Answer 1

Вместо того чтобы загружать каждый каталог в отдельный RDD, можете ли вы просто использовать пустую wildcard для загрузки всех каталогов в один 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 

Count is 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)) 

Answer 2

Вы можете уменьшить на Seq, как это (конкатенации RDD сек с ++):

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, так что будьте осторожны и тестом с большей коллекцией, хотя для стандартной библиотеки я думаю, что это безопасно.

Answer 3

Вы должны использовать union при условии искровым контексте

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