如何为地理空间点匹配最近的道路地理空间线？

针对你在Palantir Foundry中遇到的点到线最近邻匹配问题，提供两种可行方案：

方案一：修复GeoSpark依赖并实现点到线最近邻 #

1. 解决Java依赖错误 #

GeoSpark需要对应的Java JAR包才能运行，需在项目的build.gradle文件中添加以下依赖（版本需与Foundry的Spark版本匹配，Spark 3.x推荐使用GeoSpark 1.4.1）：

dependencies {
    // 其他已有依赖...
    implementation 'org.datasyslab:geospark:1.4.1'
    implementation 'org.datasyslab:geospark-sql_2.12:1.4.1'
}

同步项目依赖后，Java classpath错误即可解决。

2. 优化GeoSpark代码实现点到线最近邻 #

原代码使用的SpatialJoinQuery是范围空间连接，无法直接获取最近邻。以下代码利用GeoSpark的空间函数结合SQL，先过滤50米内道路提升性能，再为每个点匹配最近道路：

from transforms.api import transform_df, Input, Output
from geospatial_tools import geospatial
from geospark.register import GeoSparkRegistrator
from pyspark.sql import functions as F

@geospatial()
@transform_df(
    Output("ri.foundry.main.dataset.46a58ef8-732f-4bad-9b19-8e3aab9f5d30"),
    roads=Input("ri.foundry.main.dataset.32ea817c-1f13-4295-b0a1-345ca38e64d2"),
    points=Input("ri.foundry.main.dataset.e0530819-d744-49ac-9e39-91bacd41d199")
)
def compute(ctx, roads, points):
    spark = ctx.spark_session
    GeoSparkRegistrator.registerAll(spark)

    # 将Foundry Geoshape转换为GeoSpark兼容的几何类型
    points_df = points.withColumn("geom", F.expr("ST_GeomFromWKT(geoshape)"))
    roads_df = roads.withColumn("geom", F.expr("ST_GeomFromWKT(geoshape)"))

    # 注册临时视图用于SQL查询
    points_df.createOrReplaceTempView("points")
    roads_df.createOrReplaceTempView("roads")

    # 分两部分处理：优先匹配50米内最近道路，再处理无匹配的点
    result_df = spark.sql("""
        -- 第一部分：匹配50米内最近的道路
        SELECT 
            p.*,
            r.road_id AS nearest_road_id,
            ST_Distance(p.geom, r.geom) AS distance_to_road
        FROM points p
        LATERAL VIEW (
            SELECT road_id, geom
            FROM roads
            WHERE ST_Distance(p.geom, geom) <= 50
            ORDER BY ST_Distance(p.geom, geom) ASC
            LIMIT 1
        ) r

        UNION ALL

        -- 第二部分：为50米外无匹配的点全局找最近道路
        SELECT 
            p.*,
            r.road_id AS nearest_road_id,
            ST_Distance(p.geom, r.geom) AS distance_to_road
        FROM points p
        INNER JOIN (
            SELECT 
                p2.point_id,
                FIRST(r2.road_id) OVER (PARTITION BY p2.point_id ORDER BY ST_Distance(p2.geom, r2.geom) ASC) AS road_id,
                FIRST(ST_Distance(p2.geom, r2.geom)) OVER (PARTITION BY p2.point_id ORDER BY ST_Distance(p2.geom, r2.geom) ASC) AS distance_to_road
            FROM points p2
            CROSS JOIN roads r2
            WHERE NOT EXISTS (
                SELECT 1 FROM roads r3 WHERE ST_Distance(p2.geom, r3.geom) <= 50
            )
        ) r ON p.point_id = r.point_id
    """)

    # 清理临时几何列，保留原始数据和结果列
    return result_df.drop("geom")

注意：将代码中的point_id（点唯一ID）、road_id（道路ID）替换为你数据集的实际列名。

方案二：使用Palantir原生Geospatial Tools实现（无需额外依赖） #

利用原生工具的distance_join结合窗口函数，优先处理50米内的点，再针对少量无匹配的点全局查询，兼顾性能和准确性：

from transforms.api import transform_df, Input, Output
from geospatial_tools import geospatial
from pyspark.sql import Window
from pyspark.sql import functions as F

@geospatial()
@transform_df(
    Output("ri.foundry.main.dataset.46a58ef8-732f-4bad-9b19-8e3aab9f5d30"),
    roads=Input("ri.foundry.main.dataset.32ea817c-1f13-4295-b0a1-345ca38e64d2"),
    points=Input("ri.foundry.main.dataset.e0530819-d744-49ac-9e39-91bacd41d199")
)
def compute(ctx, roads, points):
    # 定义窗口：按点ID分组，按距离升序排序
    point_window = Window.partitionBy("point_id").orderBy(F.col("distance").asc())

    # 步骤1：关联50米内的道路并计算距离
    near_roads = points.distance_join(
        roads,
        points_geoshape_col="geoshape",
        other_geoshape_col="geoshape",
        max_distance=50
    ).withColumn(
        "distance",
        F.expr("ST_Distance(points_geoshape, roads_geoshape)")
    )

    # 步骤2：取每个点50米内最近的道路
    nearest_near = near_roads.withColumn(
        "rank",
        F.row_number().over(point_window)
    ).filter(F.col("rank") == 1).drop("rank", "distance", "points_geoshape", "roads_geoshape")

    # 步骤3：找出50米内无匹配的点
    no_match_points = points.join(
        nearest_near.select("point_id"),
        on="point_id",
        how="left_anti"
    )

    # 步骤4：为无匹配点全局找最近道路
    final_df = nearest_near
    if no_match_points.count() > 0:
        global_join = no_match_points.crossJoin(roads).withColumn(
            "distance",
            F.expr("ST_Distance(no_match_points.geoshape, roads_geoshape)")
        )
        nearest_global = global_join.withColumn(
            "rank",
            F.row_number().over(point_window)
        ).filter(F.col("rank") == 1).drop("rank", "distance", "no_match_points.geoshape", "roads_geoshape")
        final_df = final_df.unionByName(nearest_global)

    # 重命名道路ID列为目标名称
    return final_df.withColumnRenamed("roads_road_id", "nearest_road_id")

注意：替换代码中的point_id、road_id为实际列名，确认max_distance的单位与你的Geoshape坐标系单位一致（默认米）。

方案对比 #

• 方案一：适合超大数据量的全局最近邻查询，性能更优，但需要配置GeoSpark依赖，对环境有要求。
• 方案二：无需额外依赖，配置简单，利用50米内的优化逻辑，在大部分点符合该条件的场景下性能足够。

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