GeoSpark执行空间相交Join报错：Number of partitions must be >=0 如何解决？

执行两个含几何信息表的相交操作时，出现如下错误：

IllegalArgumentException: Number of partitions must be >= 0

使用的JOIN脚本：

tableA.
join(tableB, expr("ST_Intersects(geom, point)")).
show

表A处理逻辑与数据 #

表A有数百万行数据，处理逻辑：

spark.table("ta").
withColumn("point", expr("ST_Point(CAST(lon AS Decimal(24,20)), CAST(lat AS Decimal(24,20)))"))

处理后数据示例：

+-----------+-----------+--------------------+
|        lat|        lon|               point|
+-----------+-----------+--------------------+
|  44.978577|  30.172431|POINT (30.172431 ...|
|  44.707343|  30.794019|POINT (30.794019 ...|
|  44.817301|  30.704576|POINT (30.704576 ...|
|  44.710767|  30.657547|POINT (30.657547 ...|
|   44.88699|  30.521111|POINT (30.521111 ...|
|     44.779|    30.6296|POINT (30.6296 55...|
|  44.653987|  30.572032|POINT (30.572032 ...|
|  44.763931|  30.601646|POINT (30.601646 ...|
|44.44440079|30.50870132|POINT (30.5087013...|
|  44.707493|  30.575095|POINT (30.575095 ...|
|  44.566665|   30.56598|POINT (30.56598 5...|
|   44.58322|  30.209977|POINT (30.209977 ...|
|  44.687525|  30.665842|POINT (30.665842 ...|
|44.90000153|30.62870026|POINT (30.6287002...|
|   44.85094|  30.560021|POINT (30.560021 ...|
|   44.83429|   30.49514|POINT (30.49514 5...|
|  44.740523|  30.890627|POINT (30.890627 ...|
|  44.544804|  30.328373|POINT (30.328373 ...|
|   44.46986|    30.5456|POINT (30.5456 55...|
|    44.8912|    30.6089|POINT (30.6089 55...|
+-----------+-----------+--------------------+

表B处理逻辑与数据 #

表B仅1行数据，处理逻辑：

spark.table("tb").
withColumn("geom", expr("ST_GeomFromWKT(wkt)"))

处理后数据示例：

+--------------------+--------------------+
|                 wkt|                geom|
+--------------------+--------------------+
|MULTIPOLYGON (((3...|MULTIPOLYGON (((3...|
+--------------------+--------------------+

错误原因 #

这个异常是因为Spark空间JOIN依赖基于几何范围的分区策略，而表B只有1行数据时，默认分区数可能被优化为0，导致Spark无法计算合法的空间分区范围，进而抛出分区数非法的错误。此外，若表B的几何数据无效，也可能导致空间索引生成失败，触发同样的异常。

解决方法 #

• 强制设置表B的分区数：对表B执行repartition(1)，确保它至少有1个分区，让Spark能正常计算空间分区：

  val tableB = spark.table("tb").withColumn("geom", expr("ST_GeomFromWKT(wkt)")).repartition(1)
  

• 使用广播JOIN：由于表B数据量极小，适合用广播JOIN绕过空间分区计算。将表B广播到所有节点后，直接逐行匹配相交条件：

  import org.apache.spark.sql.functions.broadcast
  tableA.join(broadcast(tableB), expr("ST_Intersects(geom, point)")).show
  

• 验证几何数据合法性：检查表B的MULTIPOLYGON是否为有效格式，无效几何会导致分区计算失败：

  tableB.select(expr("ST_IsValid(geom)")).show
  

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