Scala（Databricks）中统计各多边形内点数量的实现方法

我来帮你搞定这个空间关联统计的问题！在Databricks的Scala环境里，要解决点和多边形的匹配问题，核心是先把字符串格式的空间数据转换成Spark能识别的几何对象，再用空间判断函数完成匹配。下面是一步步的实现方案和代码示例：

1. 导入必要的依赖与函数 #

首先要导入Spark SQL的空间处理函数，这些是完成空间判断的核心工具：

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

2. 预处理用户点数据 #

我们需要把用户的经纬度转换成Spark支持的Point几何对象，同时处理日期格式并过滤出你需要的时间段：

// 假设你的用户点DataFrame名为userPointsDF
val userPointsWithGeomDF = userPointsDF
  // 解析日期（适配你数据中的"1/1/18"格式）
  .withColumn("date_parsed", to_date(col("date"), "M/d/yy"))
  // 将经度、纬度转换为Point几何对象（注意参数顺序：经度在前，纬度在后）
  .withColumn("user_point", st_point(col("longitude"), col("latitude")))
  // 过滤特定时间段，这里示例为2018年1月1日至1月5日
  .filter(col("date_parsed").between(to_date(lit("2018-01-01")), to_date(lit("2018-01-05"))))

3. 预处理门店多边形数据 #

你的门店多边形是WKT格式的字符串（POLYGON((x y, ...))），我们需要把它转换成Spark能识别的Polygon几何对象，同时广播这个小数据集来提升关联性能：

// 假设你的门店DataFrame名为storePolygonsDF
val storesWithGeomDF = storePolygonsDF
  // 直接解析WKT格式的多边形字符串为几何对象
  .withColumn("store_polygon", st_geomFromText(col("polygon")))
  // 广播门店数据（仅1000条，大幅提升cross join的性能）
  .hint("broadcast")

4. 关联数据并统计到访次数 #

通过cross join关联两个数据集，用st_contains函数判断用户点是否落在门店多边形内，最后按门店分组统计次数：

val storeVisitCountDF = userPointsWithGeomDF
  // 关联用户点和门店数据
  .crossJoin(storesWithGeomDF)
  // 判断用户点是否在门店多边形范围内
  .filter(st_contains(col("store_polygon"), col("user_point")))
  // 按门店名称分组，统计到访次数
  .groupBy("location_name")
  .agg(count("ID").alias("visit_count"))
  // 可选：按到访次数降序排序
  .orderBy(desc("visit_count"))

5. 查看结果 #

执行以下代码查看最终的门店到访次数统计：

storeVisitCountDF.show()

关键注意事项 #

• 日期解析：如果你的日期格式和示例不同，需要调整to_date函数的格式参数（比如"MM/dd/yyyy"）。
• 空间函数兼容性：确保你的Databricks Runtime版本在7.0及以上，这些版本原生支持st_*系列空间函数。
• 性能优化：广播门店数据是关键，因为只有1000条数据，能避免大表关联的性能瓶颈。如果用户点数据量极大，可以考虑按地理分区进一步优化。

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