Spark程序的S3减慢异常[重复]

Question

我在 EMR 集群中运行了一个简单的 spark 程序，试图将 60 GB 的 CSV 文件转换为 parquet。当我提交作业时，我得到以下异常。

391, ip-172-31-36-116.us-west-2.compute.internal, executor 96): org.apache.spark.SparkException: Task failed while writing rows.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:285)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:197)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:196)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)Caused by: com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.services.s3.model.AmazonS3Exception: Slow Down (Service: Amazon S3; Status Code: 503; Error Code: 503 Slow Down; Request ID: D13A3F4D7DD970FA; S3 Extended Request ID: gj3cPalkkOwtaf9XN/P+sb3jX0CNHu/QF9WTabkgP2ISuXcXdbvYO1Irg0O54OCvKlLz8WoR8E4=), S3 Extended Request ID: gj3cPalkkOwtaf9XN/P+sb3jX0CNHu/QF9WTabkgP2ISuXcXdbvYO1Irg0O54OCvKlLz8WoR8E4=
at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1639)

Answer 1

503 Slow Down 是当您每秒执行过多请求时来自 AWS 服务的通用响应。

可能的解决方案：

1. 先将文件复制到 HDFS。
2. 您有一个 60 Gb 的文件还是有很多文件总和为 60 Gb？如果您有很多小文件，请先尝试合并它们。
3. 如果可以，请尝试减少 Parquet 输出中的分区数。 df.repartition(100)
4. 尝试使用更少的 Spark 工作程序。 val spark = SparkSession.builder.appName("Simple Application").master("local[1]").getOrCreate()

Answer 2

我很惊讶事情失败了；当 Apache s3a 客户端看到这样的问题时，它会后退：你的工作已经完成，只是更慢了。

谢尔盖的所有建议都是好的。我将从合并小文件和减少工作人员开始：较小的集群可以提供更高的性能，并节省资金。

还有一点：如果您使用 SSE-KMS 加密数据，访问该密钥也可以触发节流事件；在所有尝试使用 KMS 存储的应用程序之间共享限制。