如何在Java版Spark中使用JTS？基于geomesa-spark-jts的实现方案

1. 依赖配置 #

首先在项目中引入geomesa-spark-jts依赖，注意和你的Spark、Scala版本匹配（例如Spark 3.3.x可搭配GeoMesa 3.4.x+）：

Maven示例：

<dependency>
    <groupId>org.locationtech.geomesa</groupId>
    <artifactId>geomesa-spark-jts_2.12</artifactId>
    <version>3.5.0</version>
</dependency>

Gradle示例：

implementation 'org.locationtech.geomesa:geomesa-spark-jts_2.12:3.5.0'

2. Java中初始化JTS函数与类型 #

GeoMesa的initJts是Scala实现的单例方法，但Java可直接调用Scala单例对象（通过MODULE$）完成初始化，无需额外封装：

import org.apache.spark.sql.SparkSession;
import org.locationtech.geomesa.spark.jts.JTS;

public class SparkJtsInit {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("JavaSparkJTS")
                .master("local[*]")
                .getOrCreate();

        // 初始化JTS空间函数与类型
        JTS$.MODULE$.initJts(spark);
    }
}

如果使用JavaSparkContext，也可以这样初始化：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.locationtech.geomesa.spark.jts.JTS;

public class SparkJtsInitWithContext {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("JavaSparkJTS").setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JTS$.MODULE$.initJts(sc);
    }
}

3. JTS函数使用示例 #

初始化完成后，可直接在DataFrame或SQL中调用GeoMesa注册的空间函数（如ST_Point、ST_Distance、ST_Contains等）：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.locationtech.geomesa.spark.jts.JTS;

public class SparkJtsUsage {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("JavaSparkJTSUsage")
                .master("local[*]")
                .getOrCreate();

        JTS$.MODULE$.initJts(spark);

        // 创建测试数据
        String data = "id,x,y\n1,116.39,39.91\n2,120.16,30.24";
        Dataset<Row> df = spark.read().option("header", "true").csv(spark.createDataset(data.split("\n"), spark.implicits().stringEncoder()));

        // 使用空间函数处理数据
        df.createOrReplaceTempView("points");
        Dataset<Row> result = spark.sql("""
                SELECT id, 
                       ST_AsText(ST_Point(x, y)) AS wkt_geom,
                       ST_Distance(ST_Point(x, y), ST_Point(116.39, 39.91)) AS distance_to_beijing
                FROM points
                """);

        result.show();
    }
}

4. 注意事项 #

• 严格匹配GeoMesa、Spark、Scala的版本（例如Scala 2.12对应Spark 3.x，GeoMesa的artifactId后缀需与Scala版本一致）。
• GeoMesa会自动引入JTS核心依赖，无需单独添加jts-core等包。
• 空间函数命名与PostGIS基本一致，熟悉PostGIS的话可直接复用经验。

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