DataProc Cluster FetchFailedException [关闭]答案

【问题标题】：DataProc Cluster FetchFailedException [closed]DataProc Cluster FetchFailedException [关闭]
【发布时间】：2021-11-27 19:54:35
【问题描述】：

我会尽量使我的解释和设计参数尽可能彻底和简洁。因此，我已经进行了一些初始迭代，但我根本不擅长 Java（我最擅长的编码语言是 Python），或者不了解如何设置集群的体系结构，这样它就不会挂起或失败.在高层次上，我有一个巨大的数据集（大约 1.8 万亿个数据点，120 TB 的数据），其中包含 Lat、Lon 形式的位置数据。我正在使用 Apache Sedona 或 GeoSpark（努力了解如何在我的 python Pyspark 代码中配置和使用它们）

我的工作流程：

创建 DataProc 集群
从 GCS 存储桶和 BigQuery 表中加载数据（原始数据和一些参考数据集）
进行一些地理空间处理以提取点（例如，使用纬度、经度来分配美国州和美国县）
将新数据保存到 GCS 存储桶

所以我使用大约几百个点的小数据样本完成了这项工作。并且能够很好地做到这一点。当我尝试运行整个数据集时，它似乎遇到了很多问题。

以下只是我在 DataProc 作业日志中看到的一些内容：

WARN org.apache.spark.sql.execution.datasources.SharedInMemoryCache: Evicting cached table partition metadata from memory due to size constraints (spark.sql.hive.filesourcePartitionFileCacheSize = 262144000 bytes). This may impact query planning performance.
WARN org.apache.spark.scheduler.cluster.YarnSchedulerBackend$YarnSchedulerEndpoint: Requesting driver to remove executor 625 for reason Container marked as failed: container_1633477513137_0001_01_000626 on host: 
ERROR org.apache.spark.scheduler.cluster.YarnScheduler: Lost executor 625
WARN org.apache.spark.scheduler.TaskSetManager: Lost task 0.0 in stage 13.0 (TID 85581) (patino-pyspark-testing-sw-r96f.[<b>removed google info here</b>].internal executor 443): FetchFailed(BlockManagerId(598...
org.apache.spark.shuffle.FetchFailedException: Failed to connect to ....
Caused by: org.apache.spark.network.client.ChunkFetchFailureException: Failure while fetching StreamChunkId[streamId=493540200000,chunkIndex=0]:

这些错误我可以持续几天，我觉得这一切都归结为我不完全了解我不知道我在做什么的 Hadoop 和 Spark 配置。

我正在采用这些方法，因为我的 BQ 操作在尝试在那里处理时超时。

我真的会很好地解释以下内容：

传递 JAR 文件的正确方法（您是在作业级别传递它们还是在创建集群时传递它们？）

如何正确安装/设置 Sedona 或 GeoSpark，并不关心使用哪一个，只是想让它工作

任何和所有设置/配置（仍然是新手，所以我对后续问题表示歉意）并且我是否在创建集群或提交作业时再次传递这些

对任何日志进行故障排除。

我知道这是一堵文字墙，我非常感谢所有帮助我的努力和 cmets。再次感谢！

【问题讨论】：

@BenWatson 考虑到我的集群位于 dataproc 中，我究竟做了什么。我知道如何访问集群和作业的日志，但不确定如何在容器级别查看日志

标签： apache-spark hadoop pyspark google-cloud-dataproc

【解决方案1】：

虽然很遗憾，您关于集群架构的更一般性问题可能超出了 StackOverflow 问答的范围，但还是有一些考虑因素：

为了调试您的 Spark 作业，访问运行时信息的另一种方法是通过 Spark 的 UI 和 HistoryServer，这将提供一些工作人员级别的日志、花费的时间、发送到不同工作人员的数据量，甚至堆栈跟踪。访问 Spark UI 的最佳方式是 enable Component Gateway at cluster creation time
您的日志消息提到主机名“patino-pyspark-testing-sw-r96f” - 这似乎是默认使用Preemptible VMs 的“辅助工作者”。尽管 Dataproc 尽最大努力使典型工作负载使用这些 VM 类型尽可能平稳地运行，但最终抢占式 VM 在设计上有点难以预测。您的虚拟机很可能只是被另一个按需工作负载抢占了，从而导致临时故障。一些工作负载可以很好地处理工作人员故障的自动重试（尤其是仅映射作业和具有最少外部依赖的作业），而其他工作负载则更敏感（如果 BigQuery 依赖项使得任务级重试变得更加困难，如果您有很多连接/随机播放数据等）。如果是第一次尝试启动工作负载，您可能希望坚持使用按需 VM 类型，并且仅在您知道自己的工作可以很好地容忍任务失败时才引入 PVM。
如果 spark 确实需要洗牌/分组/聚合或缓存/检查点中间数据，它需要磁盘空间。此外，GCE VM 的 IO 性能会根据磁盘大小进行调整，因此即使您不使用大量磁盘，微型磁盘也会非常慢。如果 Spark 要使用磁盘，您可能需要确保集群中的磁盘总量至少是输入数据集大小的两倍 (120TB)。工作人员日志将提供一些指示，说明您是否有可能因“磁盘不足”错误而失去工作人员。

【讨论】：

【解决方案2】：

FetchFailedException 是一个非常常见的错误，通常发生在具有抢占式虚拟机或自动缩放但未启用 EFM 的集群中，有关更多详细信息，请参阅此doc。为避免此问题，请避免使用 PVM 或自动缩放；或在启用 PVM 或自动缩放时启用 EFM。请注意，EFM 可用于 1.4 和 1.5，但截至 2021 年 10 月，2.0 尚不可用。因此，如果您必须在 2.0 上使用 Spark 3，则必须使用第一个选项。

此外，您还需要确保集群有足够的磁盘。对于 120TB 的输入数据，我会考虑由 50 个主要工作节点组成的集群形状，每个节点都有 4TB 磁盘。

【讨论】：