Spark 向/从 S3 写入/读取 - 分区大小和压缩

Question

我正在做一个实验，以了解哪种文件大小最适合 s3 和 [EMR + Spark]

输入数据：

Incompressible data: Random Bytes in files 
Total Data Size: 20GB  
Each folder has varying input file size: From 2MB To 4GB file size.

集群规格：

1 master + 4 nodes : C3.8xls
--driver-memory 5G \
--executor-memory 3G \
--executor-cores 2 \
--num-executors 60 \

代码：

scala> def time[R](block: => R): R = {
          val t0 = System.nanoTime()
          val result = block    // call-by-name
         val t1 = System.nanoTime()
          println("Elapsed time: " + (t1 - t0) + "ns")
          result
      }
time: [R](block: => R)R

scala> val inputFiles = time{sc.textFile("s3://bucket/folder/2mb-10240files-20gb/*/*")};
scala> val outputFiles = time {inputFiles.saveAsTextFile("s3://bucket/folder-out/2mb-10240files-20gb/")};

观察

• 2MB - 32MB：大部分时间花在打开文件句柄上[效率不高]
• 64MB 直到 1GB：Spark 本身正在为所有这些文件大小启动 320 个任务，它不再是 20GB 存储桶中的文件数 数据例如512 MB 的文件有 40 个文件来制作 20gb 的数据，并且可以 只有 40 项任务要完成，但有 320 项
  每个任务处理 64MB 数据。
• 4GB 文件大小：0 字节输出 [无法处理内存/数据甚至无法拆分？？？]

问题

• 任何强制输入大小为 64MB 的默认设置？？
• 由于我使用的数据是随机字节并且已经被压缩，它如何进一步拆分这些数据？如果它可以拆分此数据，为什么它不能拆分 4gb 目标文件的文件大小 尺寸？
• 为什么通过spark上传后压缩文件变大了？ 2MB 的压缩输入文件在输出存储桶中变为 3.6 MB。

Answer 1

由于未指定，我假设在我的回答中使用 gzip 和 Spark 2.2。

• 任何强制输入大小为 64MB 的默认设置？

是的，有。 Spark 是一个 Hadoop 项目，因此将 S3 视为基于块的文件系统，即使它是基于对象的文件系统。 所以这里真正的问题是：你使用的是哪个 S3 文件系统的实现（s3a，s3n）等。类似的问题可以在here 找到。

• 由于我使用的数据是随机字节并且已经被压缩，它如何进一步拆分这些数据？如果它可以拆分此数据，为什么它不能拆分 4gb 目标文件大小的文件？

Spark docs表示可以读取压缩文件：

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

这意味着您的文件很容易被读取并转换为每一行的纯文本字符串。

但是，您使用的是压缩文件。假设是gzip等不可分割的格式，解压需要整个文件。您正在使用 3gb 执行器运行，它可以很好地满足 4mb-1gb 文件的需求，但不能一次处理大于 3gb 的文件（考虑到开销后可能会更少）。

可以在question 中找到更多信息。可拆分压缩类型的详细信息可以在此answer 中找到。

• 为什么通过spark上传后压缩文件变大了？2MB的压缩输入文件在输出桶中变成了3.6MB。

作为前一点的推论，这意味着 spark 在读取为明文时已对 RDD 进行了解压缩。重新上传时，它不再被压缩。要进行压缩，可以将压缩编解码器作为参数传递：

sc.saveAsTextFile("s3://path", classOf[org.apache.hadoop.io.compress.GzipCodec])

还有其他可用的压缩格式。