GeoMesa Spark空间Join和聚合

2021-05-06

GeoNames是一个地理信息数据库，包括一千万的地理名称和超过九百万的特征。针对纽约出租车的数据信息，假设有如下问题：出租车上下客是否集中在某些兴趣点附近？为了找到答案，需要把GeoNames数据和NYCTaxi数据结合，在结果上做聚合统计。需要Join在彼此可容许的距离之内的点，以下将其称为D-within Join。

（一）开发任务

使用GeoMesa Spark
通过地理空间的UDF创造和使用DF。
根据阈值距离计算汇总统计数据。
创造一个新的简单特征类型来表示聚合。
将结果写入data store。

（二）需求背景

（三）代码分析

$ bin/spark-shell --jars geomesa-accumulo-spark-runtime_2.11-2.0.0.jar
import org.apache.spark.sql.SparkSession
import org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator
import org.locationtech.geomesa.spark.jts._
//序列化并注册UDT/UDF，config选项也可以在conf/spark-defaults.conf中设置。
val spark: SparkSession = SparkSession.builder()
    .appName("testSpark")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.registrator", classOf[GeoMesaSparkKryoRegistrator].getName)
    .master("local[*]")
    .getOrCreate()
.withJTS
//连接到data store（本例中是Accumulo），之后将数据从store加载到DF
val taxiParams = Map(
  "accumulo.instance.id" -> "instance",
  "accumulo.zookeepers"  -> "zoo1:2181,zoo2:2181,zoo3:2181",
  "accumulo.user"        -> "user",
  "accumulo.password"    -> "password",
  "accumulo.catalog"     -> "nyctaxi")
val geonamesParams = Map(
  "accumulo.instance.id" -> "instance",
  "accumulo.zookeepers"  -> "zoo1:2181,zoo2:2181,zoo3:2181",
  "accumulo.user"        -> "user",
  "accumulo.password"    -> "password",
  "accumulo.catalog"     -> "geonames")
val taxiDF = spark.read.format("geomesa")
  .options(taxiParams)
  .option("geomesa.feature", "nyctaxi-single")
  .load()
val geonamesDF = spark.read.format("geomesa")
  .options(geonamesParams)
  .option("geomesa.feature", "geonames")
  .load()
import spark.implicits._
import org.apache.spark.sql.functions._
val geonamesNY = geonamesDF.where($"admin1Code" === lit("NY"))
// D-within Join st_bufferPoint输入点和距离（米），输出一个以该点为圆心以该距离为半径的圆。
val joinedDF = geonamesNY
  .select(st_bufferPoint($"geom", lit(50)).as("buffer"), $"name", $"geonameId")
  .join(taxiDF, st_contains($"buffer", $"pickup_point"))
//到此已经得到了纽约的景点位置以及该点的上车信息。下面分组聚合，按景点分组，并对去重后的
//上车数量计数。结果可以基于上车密度生成热图。
val aggregateDF = joinedDF.groupBy($"geonameId")
  .agg(first("name").as("name"),
       countDistinct($"trip_id")).as(s"numPickups"),
       first("buffer").as("buffer"))
//持久化存储，写入data store。先创建一个SimpleFeatureType。
import org.locationtech.geomesa.utils.geotools.SchemaBuilder
val aggregateSft = SchemaBuilder.builder()
    .addString("name")
    .addInt("numPickups")
    .addPolygon("buffer")
.build("aggregate")
//创建data store的schema
import org.geotools.data.DataStoreFinder
DataStoreFinder.getDataStore(taxiParams).createSchema(aggregateSft)
aggregateDF.write.format("geomesa").options(taxiParams)
.option("geomesa.feature", "aggregate").save()
//可视化（略）