如何用sparklyr与geospark转换Spark DataFrame为空间DataFrame并计算点线距离？

我有一个包含纬度和经度的Spark DataFrame（约100亿条观测数据，通过spark_read_parquet()加载），需要计算这些坐标与某条折线的距离。尝试将其转换为Spark空间DataFrame，用geospark结合sf方法完成计算，但直接用st_as_sf转换Spark DataFrame时出错：

s_df %>% 
  st_as_sf(coords = c("longitude", "latitude"),
           crs = "+proj=longlat +datum=WGS84 +ellps=WGS84 +towgs84=0,0,0")

错误信息：

Error in UseMethod("st_as_sf") : 
  no applicable method for 'st_as_sf' applied to an object of class "c('tbl_spark', 'tbl_sql', 'tbl_lazy', 'tbl')"

st_as_sf是针对本地R DataFrame的方法，不能直接用于Spark分布式DataFrame。要处理Spark中的空间数据，需用geospark提供的分布式空间函数来转换和计算，步骤如下：

1. 将Spark DataFrame的经纬度转换为空间点类型 #

用geospark的st_point函数在Spark中创建点几何列，注意GeoSpark中st_point的参数顺序是(longitude, latitude)：

# 转换Spark DF为空间DataFrame（添加点几何列）
s_spatial_df <- s_df %>%
  mutate(geom = st_point(longitude, latitude)) %>%
  st_set_crs(4326)  # 设置坐标系为WGS84

2. 将折线数据上传到Spark并转换为空间类型 #

本地的sf折线对象需要先转换为Spark可识别的空间格式，用geospark函数处理：

# 将本地sf折线转换为Spark空间DataFrame
polyl_spark <- polyl_sf %>%
  sdf_copy_to(sc, ., overwrite = TRUE) %>%
  mutate(geom = st_geomfromwkt(st_astext(geom))) %>%  # 转换为Spark空间几何类型
  st_set_crs(4326)

3. 计算点到折线的距离 #

使用geospark的st_distance函数完成分布式计算。如果用WGS84（EPSG:4326）大地坐标系，计算出的距离单位是度，建议先转换为投影坐标系（比如UTM）来得到以米为单位的实际距离：

# （可选）转换为投影坐标系，以米为单位计算距离
s_spatial_proj <- s_spatial_df %>% st_transform(32633)  # 示例UTM Zone 33N
polyl_proj <- polyl_spark %>% st_transform(32633)

# 计算每个点到折线的距离
distance_result <- s_spatial_proj %>%
  crossJoin(polyl_proj) %>%  # 单条折线时用crossJoin即可
  mutate(distance_to_polyline = st_distance(geom, geom)) %>%
  select(uid, distance_to_polyline)

关键注意事项 #

• 处理100亿条数据时，若有多条折线，建议用st_index创建空间索引优化关联性能，避免低效的全量关联。
• GeoSpark的空间函数参数顺序可能与sf不同，比如st_point是先经度后纬度，不要搞反。
• 大地坐标系下的距离是角度单位，必须转换为投影坐标系才能得到实际物理距离（米/千米）。

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