如何将 S3 中的多个 gzip 文件读入单个 RDD？答案

【问题标题】：How to read multiple gzipped files from S3 into a single RDD?如何将 S3 中的多个 gzip 文件读入单个 RDD？
【发布时间】：2015-02-13 04:13:51
【问题描述】：

我在 S3 上存储了许多 gzip 压缩文件，每天按项目和小时组织，文件路径的模式如下：

s3://<bucket>/project1/20141201/logtype1/logtype1.0000.gz
s3://<bucket>/project1/20141201/logtype1/logtype1.0100.gz
....
s3://<bucket>/project1/20141201/logtype1/logtype1.2300.gz

由于要每天分析数据，我必须下载并解压缩属于特定日期的文件，然后将内容组装为单个 RDD。

应该有几种方法可以做到这一点，但我想知道 Spark 的最佳实践。

提前致谢。

【问题讨论】：

标签： amazon-s3 apache-spark

【解决方案1】：

Spark 用于访问 S3 的底层 Hadoop API 允许您使用 glob expression 指定输入文件。

来自the Spark docs：

Spark 的所有基于文件的输入法，包括 textFile，都支持在目录、压缩文件和通配符上运行。例如，您可以使用textFile("/my/directory")、textFile("/my/directory/*.txt") 和textFile("/my/directory/*.gz")。

因此，在您的情况下，您应该能够使用以下方式将所有这些文件作为单个 RDD 打开：

rdd = sc.textFile("s3://bucket/project1/20141201/logtype1/logtype1.*.gz")

为了记录，您还可以使用逗号分隔列表指定文件，甚至可以将其与 * 和 ? 通配符混合使用。

例如：

rdd = sc.textFile("s3://bucket/201412??/*/*.gz,s3://bucket/random-file.txt")

简而言之，它的作用是：

* 匹配所有字符串，因此在这种情况下，201412?? 下所有文件夹中的所有gz 文件都将被加载。
? 匹配单个字符，因此201412?? 将涵盖 2014 年 12 月的所有日期，例如 20141201、20141202 等。
, 让您只需将单独的文件一次加载到同一个 RDD 中，就像本例中的 random-file.txt。

关于 S3 路径的适当 URL 方案的一些说明：

如果您在 EMR 上运行 Spark，the correct URL scheme is s3://。
如果您正在运行基于 Hadoop 2.7 或更高版本构建的开源 Spark（即没有专有的 Amazon 库），s3a:// 是您的最佳选择。
s3n:// has been deprecated 在开源方面支持 s3a://。如果您在 Hadoop 2.6 或更早版本上运行 Spark，则应仅使用 s3n://。

【讨论】：

@NickChammas 我需要担心 *.gz 文件的大小吗？所有 *.gz 文件的总大小是否有上限？
@santhosh - 上限仅取决于集群的能力。 gzip 文件越大，Spark 在开始工作之前解压缩它们的时间就越长。如果您有任何其他问题，请提出一个新问题。这个答案是关于 glob 表达式，而不是 gzip。
@NickChammas 感谢尼克的回复。我创建了一个新问题，stackoverflow.com/questions/33682709/…

【解决方案2】：

注意：在 Spark 1.2 下，正确的格式如下：

val rdd = sc.textFile("s3n://<bucket>/<foo>/bar.*.gz")

那是s3n://，不是 s3://

您还需要将您的凭据作为AWS_ACCESS_KEY_ID 和AWS_SECRET_ACCESS_KEY 放入conf/spark-env.sh。

【讨论】：

使用 s3n 和 s3 会有什么不同？我在 1.2.0 中使用过 s3，它仍在工作。任何性能改进？
@Stephane 也许我的驱动程序设置不同，但是使用开箱即用的 Spark，s3:// 不起作用，但 s3n://（通过 HDFS）可以。
如果您在 EMR 上运行 Spark，s3:// 是正确的使用方案。否则，s3n:// 已被弃用，取而代之的是 s3a://。我相应地更新了my answer。

【解决方案3】：

在 RStudio 中使用带有 Spark 2.0.0 和 SparkR 的 AWS EMR 我已经使用以下命令读取了存储在 S3 中的 gz 压缩维基百科统计文件：

df <- read.text("s3://<bucket>/pagecounts-20110101-000000.gz")

同样，对于“2011 年 1 月”下的所有文件，您可以使用上述命令，如下所示：

df <- read.text("s3://<bucket>/pagecounts-201101??-*.gz")

查看 SparkR API 文档了解更多方法。 https://spark.apache.org/docs/latest/api/R/read.text.html

【讨论】：