EMR/Spark导致S3写入速度极慢

Question

我写信是想知道有没有人知道如何加速Spark在EMR上的S3写入时间？

我的Spark作业需要超过4小时才能完成，但集群仅在前1.5小时内负载较高。

我很好奇Spark这段时间在做什么。我查看了日志，并发现有许多s3 mv命令，每个文件都有一个。直接在S3上查看后，我发现所有文件都在一个_temporary目录中。

此外，我担心我的集群成本，因为看起来我需要购买2小时的计算时间来完成此特定任务。然而，我最终购买了多达5小时的计算时间。我想知道在这种情况下EMR AutoScaling是否可以帮助降低成本。

一些文章讨论更改文件输出提交者算法，但我尝试过程中并没有取得太大成功。

sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.algorithm.version", "2")

将数据写入本地HDFS速度很快。我想知道发出一个hadoop命令将数据复制到S3是否更快？

- jspooner

---

我在这个问题上找到了两篇很棒的文章：https://hortonworks.github.io/hdp-aws/s3-spark/ 和 https://hortonworks.github.io/hdp-aws/s3-performance/index.html。 - jspooner

@JaspinderVirdee 将您的数据写入本地HDFS目录，然后使用s3-dist-cp将数据复制回S3。另外，如果您的EMR集群缺少s3-dist-cp命令，则必须在create-cluster命令中列出Hadoop。例如：--applications Name=Hadoop Name=Spark Name=Ganglia Name=zeppelin。 - jspooner

@jspooner：我遇到了同样的问题，使用s3-dist-cp工具时出现“503减速异常”。 - ljofre

@ljofre 我还没有遇到过这种情况。你的集群有多少个节点？你要传输多少数据？ - jspooner

@jspooner 我的集群有21个节点，每个节点都是r3.8xlarge类型，并且有100 GB的Parquet文件。 - ljofre

Answer 1

您看到的是输出提交器和S3之间的问题。 提交作业在临时文件夹上应用fs.rename，由于S3不支持重命名，这意味着一个请求现在正在将所有文件从_temporary复制并删除到其最终目标。

sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.algorithm.version", "2")仅适用于Hadoop版本> 2.7，它会在提交任务而不是提交作业时从临时文件夹中复制每个文件，因此分布式处理速度比较快。

如果您使用较旧版本的Hadoop，建议使用Spark 1.6并使用：

sc.hadoopConfiguration.set("spark.sql.parquet.output.committer.class","org.apache.spark.sql.parquet.DirectParquetOutputCommitter")

注意，它不能在启用speculation模式或以追加模式写入时工作。

另请注意，它在Spark 2.0中已被弃用（由algorithm.version=2替换）。

顺便说一下，在我的团队中，我们实际上使用Spark将文件写入HDFS，并在生产环境中使用DISTCP作业（特别是s3-dist-cp）将文件复制到S3，但这是出于其他几个原因（一致性，容错性），所以并不必要。您可以使用我建议的方法很快地将文件写入S3。

- Tal Joffe

---

更快但风险更高。Spark 2不仅删除了文件，如果您使用任何带有“direct”一词的提交者，您将被告知并禁用猜测。 - stevel

啊，这对我不起作用，因为我正在使用附加模式！ - jspooner

@SteveLoughran，阅读了您的回答后，我明白了您的意思。我可能会尝试您的建议来替换当前的解决方案。 - Tal Joffe

s3-dist-cp 在 HDFS 上创建临时文件，成功后未能删除，导致我失去了空间，是否有参数可以告诉 s3-dist-cp 删除临时文件。 - loneStar

@Achyuth 抱歉，它删除了源文件。没有注意到问题...当复制成功时，临时文件会移动到最终目的地，而不管--deleteOnSuccess标志如何。你可能仍然会留下空文件夹，但我认为这不是什么值得关注的事情。在某些情况下，我们有一个Python脚本，每天删除所有大小为0的文件夹以清理这些文件夹。 - Tal Joffe

Answer 2

我有一个类似的用例，其中我使用spark写入s3并遇到了性能问题。主要原因是spark创建了大量的零字节部分文件，并且将临时文件替换为实际文件名会减慢写入过程。我尝试了以下方法作为解决方法：

1. 将spark的输出写入HDFS，并使用Hive将其写入s3。性能要好得多，因为Hive创建的part文件数量较少。我遇到的问题是（在使用spark时也遇到了相同的问题），由于安全原因，在生产环境中不提供策略上的删除操作。在我的情况下，S3存储桶是KMS加密的。

2. 将spark输出写入HDFS，将hdfs文件复制到本地，并使用aws s3 copy将数据推送到s3。这种方法的效果次之。我向亚马逊提交了工单，他们建议采用这种方法。

3. 使用s3 dist cp将文件从HDFS复制到S3。这个方法没有问题，但不够高效。

- vikrame

---

亚马逊可能最近对s3-dist-cp进行了改进。在我们的EMR集群上，它表现出色，将约200GB的数据从HDFS复制到S3只需不到2分钟。 - Jason Evans

@vikrame：请问您能告诉我如何将文件保存在本地HDFS上，然后复制到S3吗？ - Xi12

Answer 3

"直接提交器"已从Spark中移除，因为它不具备故障恢复能力。我强烈建议不要使用它。

Hadoop正在进行工作，添加0重命名提交器，这将是O(1)和容错的；请关注 HADOOP-13786。

暂时忽略“魔术提交器”，基于Netflix的分段提交器将会首先发布（hadoop 2.9？3.0？）

1. 在任务提交时将工作写入本地FS
2. 发出未提交的多部分PUT操作以写入数据，但不实现它
3. 使用原始的“算法1”文件输出提交器保存提交PUT到HDFS所需的信息
4. 实现作业提交，使用HDFS的文件输出提交确定完成哪些PUT并取消哪些PUT。

结果：任务提交需要花费数据/带宽几秒钟，而作业提交不超过在目标文件夹上执行1-4个GET和每个挂起的文件的POST所需的时间，后者是并行的。

您可以从Netflix获取此工作所基于的提交者，并可能今天就在Spark中使用它。请设置文件提交算法= 1（应该是默认值），否则它实际上不会写入数据。

- stevel

---

Steve，是否有一个Apache JIRA问题来包含Netflix S3提交者在Hadoop中？我似乎找不到。 - Jonathan Kelly

这是HADOOP-13786。由于正在与S3a Phase II工作混合，因此该集成在一段时间内不会成为您可以使用的任何内容。原始代码今天应该可以工作（Netflix使用它），本周发布了Hadoop 2.8.0 RC，其中包含所有读取和写入管道加速。 - stevel

当然，正如LI所说，您是EMR团队的一员，拥有自己的s3客户端。Netflix的Staging提交者应该可以在那里工作，“Magic”提交者绝对不行，但这已经成为次要问题，第一要务是Staging提交者。https://github.com/steveloughran/hadoop/blob/s3guard/HADOOP-13786-committer/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3a_committer.md - stevel

谢谢，史蒂夫！我只是想更多地了解那个提交者，因为我还没有听说过它，而且我想能够跟踪 JIRA 如果有的话。 - Jonathan Kelly

Answer 4

我曾经遇到同样的问题，后来找到了一种解决方案，就是改变S3协议。原本我使用的是s3a://来读写数据，然后我改成了只用s3://，结果完美解决了问题，实际上我的进程还追加了数据。

- Arturo Pineda

---

你能否提供更多关于你的方法的细节？例如使用的Spark和Hadoop版本，以及Jars。 - Chitral Verma

看起来s3://现在已经被弃用了，根据这个链接：https://cwiki.apache.org/confluence/display/HADOOP2/AmazonS3。它会抛出类未找到的错误。 - xanjay

Answer 5

如果您在Spark输出中看到大量的重命名操作，请阅读this。

- Niros

Answer 6

我们在Azure上使用Spark on WASB时遇到了同样的问题。最终我们决定不直接使用分布式存储与Spark。我们将spark.write到真正的hdfs://目的地，并开发了一个特定的工具：hadoop copyFromLocal hdfs:// wasb://。然后，HDFS是我们的临时缓冲区，在归档到WASB（或S3）之前。

- Julien Laurenceau

Answer 7

你要写入的文件有多大？如果只有一个核心在写入非常大的文件，那么速度会比将文件分割并让多个工作线程写入较小的文件要慢得多。

- Ted