DataProc集群FetchFailedException

Question

我将尽量用我的解释和设计参数做到既透彻又简洁。因此，我给出了一些初始迭代，我根本不擅长Java (我喜欢的编码语言是Python)，也不了解如何设置集群的体系结构，这样集群就不会挂起或失败。在高层次上，我有一个庞大的数据集(大约1.8万亿个数据点，120 TB数据)，这些数据集的位置数据以Lat，Lon的形式存在。我使用的是Apache或GeoSpark (很难理解如何在python代码中配置和使用它们)

我的工作流程：

• 创建DataProc集群
• 从GCS桶和BigQuery表加载数据(原始数据和一些参考数据集)
• 做一些地理空间处理来提取点(例如，使用Lat、Lon来指定美国州和美国县)
• 将新数据保存到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和火花的配置，我不知道我在做什么。

我采取这些方法，因为我的烧烤操作已超时时，试图处理那里。

我想就以下几点作一个很好的解释：

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

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

任何和所有的设置/配置(同样还是新手，所以我对后续问题表示歉意)，我是在创建集群时还是在提交作业时传递这些设置/配置？

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

我知道这是一堵墙的文字，并真正感谢任何和所有的努力和评论，以帮助我。再次感谢您！

Answer 1

虽然您关于集群体系结构的更普遍的问题可能超出了StackOverflow Q&A所能涵盖的范围，但有几个考虑因素出现在您的脑海中：

1. 对于调试Spark作业，另一种访问运行时信息的方法是通过Spark的UI和HistoryServer，这将提供一些员工级别的日志、花费的时间、发送给不同工作人员的数据量，甚至堆栈跟踪。访问Spark的最佳方法是如果您在集群创建时启用组件网关
2. 您的日志消息提到了主机名“patino-吡火花-测试-sw-r96f”--这似乎是一个默认使用可抢占VM的“辅助工作人员”。虽然Dataproc尽最大努力使典型的工作负载在这些VM类型下尽可能平稳地运行，但最终可抢占VM在设计上还是有点不可预测的。您的VM可能很容易被另一个按需工作负载抢占，从而导致临时故障。一些工作负载可以很好地处理工作人员故障的自动重试(特别是只映射具有最小外部依赖性的作业和作业)，而其他工作负载则更敏感(如果BigQuery依赖关系使任务级重试变得更困难，如果您有大量连接/洗牌数据，等等)。如果试图第一次完成工作负载，您可能希望坚持按需VM类型，并且只在知道您的工作能够很好地容忍任务失败时才引入PVM。
3. 如果spark确实需要洗牌/分组/聚合或缓存/检查点中间数据，那么它将需要磁盘空间。此外，GCE的IO性能根据磁盘大小进行调整，因此即使您不使用大量磁盘，微小的磁盘也会非常缓慢。如果Spark要使用磁盘，您可能希望确保集群中的磁盘总量至少是输入数据集(120 to )的两倍。工作日志将给出一些指示，说明是否可能因为“磁盘外”错误而丢失工作人员。

Answer 2

FetchFailedException是一个非常常见的错误，通常发生在具有可抢占VM或自动标度但没有启用EFM的集群中，有关详细信息，请参阅此文档。若要避免此问题，请避免使用PVM或自动标号；或在启用PVM或自动标号时启用EFM。请注意，EFM的版本为1.4和1.5，但截至2021年10月还不能用于2.0。因此，如果您必须在2.0上使用Spark 3，则必须选择第一个选项。

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