如何在sparklyr中高效分块运行自定义Haversine距离计算函数？

直接解决：基于分组列的spark_apply分块处理 #

你已经完成了数据分组逻辑，只需在group_by后调用spark_apply时指定grouped = TRUE，即可让Spark按分组并行处理每个数据块，避免一次性加载全量数据：

# 为全量关联数据添加分组列
full_sf_with_groups <- full_sf %>%
  sdf_with_sequential_id(id = "id", from = 1L) %>%
  mutate(equal_groups = ntile(id, 4)) # 可根据集群资源调整分组数

# 按分组分块执行距离计算
result <- full_sf_with_groups %>%
  group_by(equal_groups) %>%
  spark_apply(
    get_geodesic_distance,
    grouped = TRUE,
    # 提前指定返回Schema，避免Spark自动推断带来的性能损耗
    schema = list(
      place_x = "integer",
      long1 = "double",
      lat1 = "double",
      place_y = "character",
      long2 = "double",
      lat2 = "double",
      distance = "double",
      equal_groups = "integer"
    )
  )

# 查看结果
result %>% head()

更优方案：避免生成全量笛卡尔积 #

直接生成50亿行的笛卡尔积会占用大量存储和内存，建议先对客户数据分块，再让每个客户块与全部门店关联并计算距离，从根源减少中间数据量：

# 对客户数据分块（示例分成100组，可根据集群executor数量调整）
from_many_sf_with_groups <- from_many_sf %>%
  sdf_with_sequential_id(id = "id", from = 1L) %>%
  mutate(customer_group = ntile(id, 100))

# 定义分块处理逻辑：客户块关联门店+计算距离
process_customer_block <- function(customer_block) {
  customer_block %>%
    full_join(to_sf, by = character()) %>%
    get_geodesic_distance()
}

# 按客户分组并行处理
final_result <- from_many_sf_with_groups %>%
  group_by(customer_group) %>%
  spark_apply(
    process_customer_block,
    grouped = TRUE,
    schema = list(
      place = "integer",
      long1 = "double",
      lat1 = "double",
      place_y = "character",
      long2 = "double",
      lat2 = "double",
      distance = "double",
      customer_group = "integer"
    )
  )

性能升级：使用Spark原生UDF替代spark_apply #

spark_apply依赖R进程处理数据，存在JVM与R的序列化开销。将Haversine函数转为Spark原生UDF，性能会大幅提升：

# 注册Spark SQL原生UDF
spark_udf(sc, "haversine_distance", function(long1, lat1, long2, lat2) {
  deg2rad <- function(deg) deg * pi / 180
  # 经纬度转弧度
  long1 <- deg2rad(long1)
  lat1 <- deg2rad(lat1)
  long2 <- deg2rad(long2)
  lat2 <- deg2rad(lat2)
  
  R <- 6378137 # 地球半径（米）
  diff_long <- long2 - long1
  diff_lat <- lat2 - lat1
  
  a <- sin(diff_lat/2)^2 + cos(lat1) * cos(lat2) * sin(diff_long/2)^2
  c <- 2 * atan2(sqrt(a), sqrt(1-a))
  R * c # 返回距离（米）
})

# 直接调用UDF计算，无需spark_apply
final_result_native <- full_sf %>%
  mutate(distance = haversine_distance(long1, lat1, long2, lat2))

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