DataProc集群PySpark处理海量地理数据FetchFailedException报错及配置咨询

1. JAR文件传递方式 #

两种方式都支持，根据使用场景选择即可：

• 集群创建阶段配置：适用于所有运行在该集群上的任务都需要的公共依赖。创建DataProc集群时通过--jars参数指定GCS上的JAR路径，或者通过初始化脚本统一安装依赖，避免每个任务重复传递JAR，提升提交效率。
• 任务提交阶段配置：适用于仅当前任务需要的专属依赖。提交PySpark任务时通过spark-submit的--jars参数指定路径，也可以直接用--packages参数传入Maven坐标，由Spark自动下载对应依赖，无需手动提前下载JAR包，更适合Python开发者使用。

2. Apache Sedona PySpark 配置方法 #

推荐使用Apache Sedona（原GeoSpark），配置流程无需手动处理Java依赖，对Python开发者友好：

1. 提前安装Python端依赖：pip install apache-sedona[spark]，可以直接放到DataProc的初始化脚本中，或者任务运行前执行。
2. 提交任务时指定Sedona的Maven依赖，示例spark-submit参数：

--packages org.apache.sedona:sedona-spark-shaded-3.4_2.12:1.5.0,org.datasyslab:geotools-wrapper:1.5.0-28.2

注意选择和你集群Spark版本匹配的Sedona版本即可。
3. 代码中初始化Sedona上下文，直接注册所有空间处理UDF和序列化配置，无需手动额外设置：

from pyspark.sql import SparkSession
from sedona.spark import SedonaContext
from pyspark.sql.functions import expr

spark = SparkSession.builder.appName("geo_process").getOrCreate()
spark = SedonaContext.create(spark)

# 后续即可直接调用空间函数，比如经纬度转点、空间匹配示例
df = df.withColumn("point", expr("ST_Point(longitude, latitude)"))
# 加载美国州县边界表，广播小表提升匹配效率
boundary_df = spark.read.parquet("gcs://xxx/county_boundary.parquet")
result = df.join(boundary_df, expr("ST_Contains(boundary_geom, point)"))

3. 相关参数配置说明及生效阶段 #

集群创建阶段配置（全局生效，所有任务通用） #

• 集群规格配置：单Executor配置建议4核16G内存，开启外部Shuffle服务和动态资源分配，对应参数：

spark.shuffle.service.enabled=true
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=10
spark.dynamicAllocation.maxExecutors=200 # 根据你的数据量调整

• 分区元数据缓存配置：解决日志中分区元数据驱逐的警告，调大缓存上限：

spark.sql.hive.filesourcePartitionFileCacheSize=2147483648 # 设为2G

任务提交阶段配置（仅当前任务生效） #

• Sedona序列化配置：提升空间数据处理性能

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrator=org.apache.sedona.core.serde.SedonaKryoRegistrator

• Shuffle相关配置：解决大量数据Shuffle时的稳定性问题

spark.sql.shuffle.partitions=4000 # 120TB数据建议设为2000~5000，根据实际运行调整
spark.executor.memoryOverhead=4096 # 设为Executor内存的20%~30%
spark.sql.broadcastTimeout=3600 # 广播小表超时时间设为1小时，避免边界表广播超时
spark.shuffle.io.maxRetries=10 # 加大Shuffle拉取重试次数，避免临时网络波动导致失败

4. 日志报错通用排查方法 #

对应你提供的报错信息，按优先级排查：

1. 先处理分区元数据缓存警告：调大spark.sql.hive.filesourcePartitionFileCacheSize参数，避免查询规划性能下降导致任务运行变慢。
2. 排查Executor丢失问题：首先查看对应Executor的单独日志，若为内存溢出（OOM），则调大Executor内存或内存 overhead，也可以增大Shuffle分区数减少单个Task处理的数据量。
3. 排查FetchFailed错误：首先确认集群是否开启了外部Shuffle服务，未开启的话Executor异常退出会导致Shuffle数据丢失，直接触发该错误；其次检查是否存在数据倾斜，单个Key对应的数据量过大导致Task运行超时，Executor被Yarn回收，可以通过拆分倾斜Key、增大Shuffle分区数解决。
4. 网络稳定性排查：若上述配置都无问题，检查DataProc集群节点是否在同一可用区，VPC安全组是否放开了Spark内部通信端口，是否存在网络限流的情况。

内容的提问来源于stack exchange，提问作者cpatino08