apache-spark - 即使在 8 小时后，Spark 写入 CSV 也会失败

Question

我有一个数据框，其中包含大约 200-600 GB 的数据，我正在读取、处理这些数据，然后使用 spark shell (scala) 将其写入 csv。在弹性映射减少集群上。即使在 8 小时后，Spark 写入 CSV 也会失败

这是我写入 csv 的方式:

result.persist.coalesce(20000).write.option("delimiter",",").csv("s3://bucket-name/results")

结果变量是通过混合来自其他一些数据帧的列创建的: var result=sources.join(destinations, Seq("source_d","destination_d")).select("source_i","destination_i")

现在，我能够在大约 22 分钟内读取它所基于的 csv 数据。在同一个程序中，我还能够在 8 分钟内将另一个(较小的)数据帧写入 csv。然而，对于这个result dataframe 它需要 8 个多小时但仍然失败......说其中一个连接已关闭。

我也在 13 x c4.8xlarge instances on ec2 上运行这项工作，每个都有 36 个内核和 60 GB 的内存，所以我认为我有能力写入 csv，尤其是在 8 小时后。

许多阶段需要重试或任务失败，我无法弄清楚我做错了什么或为什么要花这么长时间。我可以从 Spark UI 看到它甚至从未进入写入 CSV 阶段并且忙于持久化阶段，但是如果没有持久化功能它在 8 小时后仍然失败。有任何想法吗？非常感谢您的帮助!

更新:

我运行了以下命令来重新分区 result可变为 66K 分区:

val r2 = result.repartition(66000) #confirmed with numpartitions
r2.write.option("delimiter",",").csv("s3://s3-bucket/results")

然而，即使在几个小时后，作业仍然失败。我还做错了什么？

请注意，我正在通过 spark-shell yarn --driver-memory 50G 运行 spark shell

更新 2:

我试过先用 persist 运行写入:

r2.persist(StorageLevel.MEMORY_AND_DISK)

但是我有很多阶段都失败了，返回一个，Job aborted due to stage failure: ShuffleMapStage 10 (persist at <console>:36) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 3'或者说 Connection from ip-172-31-48-180.ec2.internal/172.31.48.180:7337 closed

执行者页面

节点返回洗牌错误的 Spark Web UI 页面

返回 ec2 连接关闭错误的节点的 Spark Web UI 页面

整体工作摘要页面

Answer 1

I can see from the Spark UI that it never even got to the write CSV stage and was busy with persist stages, but without the persist function it was still failing after 8 hours. Any ideas?

是FetchFailedException即无法获取洗牌 block

既然你能够处理小文件，那么只有大数据才会失败......我强烈觉得分区不够。

首先是验证/打印 source.rdd.getNumPartitions()。和 destinations.rdd.getNumPartitions()。和 result.rdd.getNumPartitions()。

数据加载完成后需要重新分区，以便将数据分区(通过shuffle)到集群中的其他节点。这将为您提供更快处理而不会失败所需的并行性

此外，要验证应用的其他配置...像这样打印所有配置，根据需要将它们调整为正确的值。

sc.getConf.getAll

也看看

• SPARK-5928
• Spark-TaskRunner-FetchFailedException 可能的原因:OOM 或 Container 内存限制