Spark Sedona中Join转GroupBy任务卡滞在GET RESULT的问题求助

• 环境：Docker镜像中运行Spark 3.4.1 + Sedona + Jupyter Lab
• 任务：从5M条测试船只点位数据（全量5475M，含Lon、Lat、日期等字段）统计点位最多的月份，需通过ST_Contains关联60条多边形Shapefile数据（含multipolygon、名称字段），再按多边形名称分组统计内部点位数量
• 故障现象：执行关联分组查询时卡滞在GET RESULT状态，最终报错TaskResultLost (result lost from block manager)
• 已尝试操作：增加executor数量与内存、修改spark.shuffle.blockTransferService为nio、调整查询语句；将多边形设为左表时，单个executor运行耗时约2小时，并行化触发上述报错

1. 空间关联执行计划优化（核心） #

Sedona的ST_Contains直接关联本质是笛卡尔积+空间判断，极易引发数据倾斜或shuffle量过载，尤其是点数据量远大于多边形时：

• 改用空间分区+广播多边形的策略：先对点数据按空间范围分区，再将仅60条的多边形数据广播到每个分区，避免全量shuffle
  示例代码：

  // 提前将点转换为空间类型并缓存，避免重复计算
  val pointsWithGeom = pointsDF.withColumn("point", ST_Point($"Lon", $"Lat")).cache()
  // 按空间范围对点数据分区（可根据数据量调整分区数）
  val partitionedPoints = pointsWithGeom.repartitionByRange(100, $"point")
  // 广播多边形数据
  val broadcastPolygons = broadcast(polygonsDF)
  // 关联时仅对同分区内的点和多边形做空间判断
  val result = partitionedPoints.join(broadcastPolygons, 
    ST_Contains(broadcastPolygons("multipolygon"), partitionedPoints("point")))
    .groupBy("多边形名称")
    .count()
  

2. 针对TaskResultLost的配置调整 #

该报错多与shuffle数据传输、block管理内存或网络超时有关，结合Docker环境特性调整：

• 增大spark.driver.maxResultSize：默认值过小会导致大结果丢失，建议设为4g（根据实际内存调整）

  --conf spark.driver.maxResultSize=4g
  

• 扩大blockManager端口范围，避免Docker环境下端口冲突：

  --conf spark.blockManager.port=10000-10050
  

• 延长网络超时时间，适配Docker网络可能存在的延迟：

  --conf spark.network.timeout=300s
  

• 关闭shuffle文件合并（部分环境下开启会引发block丢失）：

  --conf spark.shuffle.consolidateFiles=false
  

3. Docker环境专属优化 #

• 确保容器分配足够的CPU和内存配额，避免executor被容器内核强制终止导致block丢失
• 采用host网络模式运行Spark容器，减少网络虚拟化带来的延迟和端口转发问题：

  docker run --network=host ...
  

• 关闭Docker的userland-proxy，避免端口转发异常：

  docker daemon --userland-proxy=false
  

4. 任务拆分减负 #

• 按月份拆分点数据，分批次统计每个月的点位后再合并结果，降低单任务处理量：

  val monthlyStats = pointsWithGeom.groupBy(month($"日期").alias("month"))
    .flatMapGroups { case (month, pointsIter) =>
      val polygons = broadcastPolygons.value
      pointsIter.flatMap(point => 
        polygons.filter(p => ST_Contains(p("multipolygon"), point("point")))
          .map(p => (month, p.getAs[String]("多边形名称")))
      )
    }
    .groupBy("month", "_2")
    .count()
  

• 提前过滤无效点位（如Lon/Lat不在有效地理范围的记录），减少无效计算

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