如何使用scala从spark中的RDD获取值答案

【问题标题】：How to get values from RDD in spark using scala如何使用scala从spark中的RDD获取值
【发布时间】：2016-03-13 00:46:14
【问题描述】：

我想读取存储在特定位置的 zipfile 的内容。所以我使用了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 实例。

【问题讨论】：

标签： scala apache-spark bigdata rdd

【解决方案1】：

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

【讨论】：

【解决方案2】：

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

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

使用 Apache Commons Compress 库，您可以执行以下操作来获取 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()

【讨论】：