Scala Spark地理空间相交问询：Polyline与Segments数据处理

嘿，咱们来搞定Scala Spark下Polyline和Segments数据集的地理空间相交问题！Spark本身没有原生的地理空间操作API，所以咱们用GeoSpark这个开源库就行，它对Spark的地理空间支持特别完善。下面是一步步的实现方案：

1. 先明确数据集结构 #

先把两个数据集的Schema列出来，方便咱们对齐逻辑：

// Polyline数据集Schema
root
 |-- Polyline_id: string (nullable = false)
 |-- latitude: double(nullable = true)
 |-- longitude: double(nullable = true)
 |-- point_index: integer(nullable = true)

// Segments数据集Schema
root
 |-- Segments_id: string (nullable = false)
 |-- latitudeStart: double(nullable = true)
 |-- longitudeStart: double(nullable = true)
 |-- latitudeEnd: double(nullable = true)
 |-- longitudeEnd: double(nullable = true)

2. 引入GeoSpark依赖 #

如果是Maven项目，先在pom.xml里添加GeoSpark的依赖（建议用最新稳定版本，这里举个示例版本）：

<dependency>
    <groupId>org.datasyslab</groupId>
    <artifactId>geospark-sql_2.12</artifactId>
    <version>1.4.1</version>
</dependency>

3. 初始化SparkSession并注册GeoSpark函数 #

必须注册GeoSpark的空间函数才能使用它的地理空间能力：

import org.apache.spark.sql.SparkSession
import org.datasyslab.geosparksql.utils.GeoSparkSQLRegistrator

val spark = SparkSession.builder()
  .appName("PolylineSegmentsIntersection")
  .master("local[*]") // 生产环境请移除该配置，用集群模式
  .getOrCreate()

// 注册GeoSpark所有SQL空间函数
GeoSparkSQLRegistrator.registerAll(spark)

4. 将数据集转换为地理几何类型 #

处理Polyline：将同ID的点排序生成LineString #

每个Polyline_id对应多个离散点，需要按point_index排序后拼接成完整的线（LineString）：

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

// 替换成你的Polyline数据读取逻辑
val polylineDF = spark.read.format("parquet").load("/path/to/polyline")

val polylineGeoDF = polylineDF
  // 按Polyline_id分组，收集所有点的信息
  .groupBy("Polyline_id")
  .agg(collect_list(struct("longitude", "latitude", "point_index")).alias("points"))
  // 按point_index从小到大排序点，保证线的顺序正确
  .withColumn("sorted_points", sort_array(col("points"), asc = true))
  // 将排序后的点拼接成WKT格式的LineString字符串
  .withColumn("line_wkt", concat(
    lit("LINESTRING("),
    concat_ws(",", expr("transform(sorted_points, p -> concat(p.longitude, ' ', p.latitude))")),
    lit(")")
  ))
  // 将WKT字符串转换为GeoSpark可识别的Geometry类型
  .withColumn("polyline_geom", st_geomfromwkt(col("line_wkt")))
  // 保留需要的字段
  .select("Polyline_id", "polyline_geom")

处理Segments：将起点终点转换为LineString #

每个Segment本身就是一条线段，直接用起点和终点生成LineString即可：

// 替换成你的Segments数据读取逻辑
val segmentsDF = spark.read.format("parquet").load("/path/to/segments")

val segmentsGeoDF = segmentsDF
  // 拼接成WKT格式的LineString字符串
  .withColumn("line_wkt", concat(
    lit("LINESTRING("),
    concat_ws(" ", col("longitudeStart"), col("latitudeStart")),
    lit(","),
    concat_ws(" ", col("longitudeEnd"), col("latitudeEnd")),
    lit(")")
  ))
  // 转换为Geometry类型
  .withColumn("segment_geom", st_geomfromwkt(col("line_wkt")))
  // 保留需要的字段
  .select("Segments_id", "segment_geom")

5. 执行空间相交操作 #

用GeoSpark的st_intersects函数判断两条线是否相交，然后关联两个数据集：

val intersectionResultDF = polylineGeoDF
  // 先做笛卡尔积，再过滤相交的记录（大数据量建议用空间索引优化）
  .crossJoin(segmentsGeoDF)
  .filter(st_intersects(col("polyline_geom"), col("segment_geom")))
  .select(
    col("Polyline_id"),
    col("Segments_id"),
    // 可选：获取相交的具体几何部分（比如交点或重叠线段）
    st_intersection(col("polyline_geom"), col("segment_geom")).alias("intersection_geom")
  )

// 查看结果
intersectionResultDF.show(false)

性能优化小贴士 #

如果数据集很大，直接crossJoin会非常耗时，建议给几何数据建立空间索引，大幅提升相交判断的效率：

// 给Polyline和Segment分别建立空间索引
val indexedPolylineDF = polylineGeoDF.withColumn("spatial_index", st_index(col("polyline_geom")))
val indexedSegmentsDF = segmentsGeoDF.withColumn("spatial_index", st_index(col("segment_geom")))

// 用索引关联替代笛卡尔积
val optimizedResultDF = indexedPolylineDF
  .join(indexedSegmentsDF, st_intersects(col("polyline_geom"), col("segment_geom")))
  .select("Polyline_id", "Segments_id", "intersection_geom")

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