如何使用scala从星火中的RDD中获取值

Question

我想阅读存储在特定位置的the文件的内容。因此，我使用了SparkContext.readFile方法，如下所示：

val zipFileRDD = sc.binaryFiles("./handsOn/repo~apache~storm~14135470~false~Java~master~2210.zip")

zipFileRDD: org.apache.spark.rdd.RDD[(String, org.apache.spark.input.PortableDataStream)] = ./handsOn/repo~apache~storm~14135470~false~Java~master~2210.zip BinaryFileRDD[4] at binaryFiles at <console>:21

我的问题是：

如何从这个RDD. PortableDataStream .中获取实例

Answer 1

您可以使用 action：zipFileRDD.collect返回一个Array[(String, PortableDataStream)]。但这通常不是你真正想要的！如果使用这些实例读取文件，那么实际上并不是在使用Spark的功能:驱动程序中会发生任何事情。相反，应用map和其他转换，以便在不同的工作人员上读取不同的文件。

Answer 2

如果您只想在RDD之外运行PortableDataStream，那么只需运行：

val zipFilePds = zipFileRDD.map(x => x._2).collect()

使用库，您可以执行类似的操作来获取zip文件的内容(在本例中是文件清单)：

import org.apache.commons.compress.archivers.zip

val zipFileListing = zipFileRDD.map(x => x._2.open()).map(x => { val y = new ZipArchiveInputStream(x) ; y.getNextEntry().getName() }).collect()