Spark中为列表元素关联经纬度坐标的Scala实现方案问询

我有如下结构的Spark Dataset：

+----------+-------+----+---+--------------+
|        _1|     _2|  _3| _4|            _5|
+----------+-------+----+---+--------------+
|      null|1111111|null| 15|       [98765]|
|      null|2222222|null| 16|[97008, 98765]|
|6436334664|3333333|null| 15|       [97008]|
|2356242642|4444444|null| 11|       [97008]|
+----------+-------+----+---+--------------+

其中第5列是该行关联的zipcode列表。另有一张zipcode坐标表，每行对应唯一zipcode及对应的经度、纬度。需要生成如下结构的Dataset：

+----------+-------+----+---+--------------+-----------------------------------
|        _1|     _2|  _3| _4|            _5|                                _6|
+----------+-------+----+---+--------------+----------------------------------+
|3572893528|1111111|null| 15|       [98765]| [(54.12,-80.53)]                 |
|5325232523|2222222|null| 16|[98765, 97008]| [(54.12,-80.53), (44.12,-75.11)] |
|6436334664|3333333|null| 15|       [97008]| [(54.12,-80.53)]                 | 
|2356242642|4444444|null| 11|       [97008]| [(54.12,-80.53)]                 |
+----------+-------+----+---+--------------+----------------------------------+

第6列是第5列zipcode按原顺序对应的坐标列表。

尝试过每次需要坐标时过滤zipcode表，出现NullPointerException；先收集zipcode表到内存再过滤又会内存溢出。使用Scala开发，通过Spark SQL获取原始Dataset，求可行方案。

采用分布式关联+按顺序聚合的方式，避免driver端内存溢出和UDF闭包导致的NPE，具体步骤如下：

1. 定义数据结构（可选，增强类型安全） #

先定义对应Dataset的case类：

// 原始数据结构
case class OriginalData(_1: Option[String], _2: String, _3: Option[String], _4: Int, _5: List[String])
// zipcode坐标表结构
case class ZipCoord(zipcode: String, lon: Double, lat: Double)
// 带索引的中间结构
case class ExplodedData(_1: Option[String], _2: String, _3: Option[String], _4: Int, zipcode: String, idx: Int)
// 关联后的中间结构
case class JoinedData(_1: Option[String], _2: String, _3: Option[String], _4: Int, idx: Int, coord: (Double, Double))
// 最终结果结构
case class ResultData(_1: Option[String], _2: String, _3: Option[String], _4: Int, _5: List[String], _6: List[(Double, Double)])

2. 展开zipcode列表并保留顺序索引 #

使用posexplode函数将原始数据的zipcode列表展开，同时保留每个zipcode在原列表中的位置索引：

import org.apache.spark.sql.functions._
import spark.implicits._

// 假设originalDs是你的原始Dataset，zipCoordsDs是zipcode坐标表Dataset
val explodedDs = originalDs
  .select($"_1", $"_2", $"_3", $"_4", posexplode($"_5").as(Seq("idx", "zipcode")))
  .as[ExplodedData]

3. 关联zipcode坐标表 #

通过zipcode字段做分布式join，获取每个zipcode对应的坐标：

val joinedDs = explodedDs
  .join(zipCoordsDs, explodedDs("zipcode") === zipCoordsDs("zipcode"), "left")
  .select($"_1", $"_2", $"_3", $"_4", $"idx", struct($"lon", $"lat").as("coord"))
  .as[JoinedData]

这里用left join可以避免因zipcode不存在导致的数据丢失，若不需要可改为inner join。

4. 按原行分组并按索引聚合坐标列表 #

按原始数据的唯一标识（_1、_2、_3、_4）分组，然后用sort_array+collect_list按索引顺序聚合坐标，保证和原zipcode列表顺序一致：

val aggregatedDs = joinedDs
  .groupBy($"_1", $"_2", $"_3", $"_4")
  .agg(
    sort_array(collect_list(struct($"idx", $"coord"))).as("sorted_coords")
  )
  .select(
    $"_1", $"_2", $"_3", $"_4",
    // 提取排序后的坐标列表
    expr("transform(sorted_coords, x -> x.coord)").as("_6")
  )

5. 关联回原始数据的zipcode列表（可选） #

如果需要保留原始的_5列，将聚合结果和原始Dataset关联：

val finalResultDs = originalDs
  .join(aggregatedDs, Seq("_1", "_2", "_3", "_4"), "inner")
  .select($"_1", $"_2", $"_3", $"_4", $"_5", $"_6")
  .as[ResultData]

方案优势 #

• 全程基于Spark分布式计算，不会将zipcode表全量加载到driver内存，避免内存溢出。
• 用posexplode+sort_array保证坐标顺序和原zipcode列表完全一致。
• 避免了在UDF中操作Dataset导致的闭包序列化问题，从根源上解决NPE。

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