GeoMesa Spark广播和聚合

GDELT提供了一个全面的时间和位置索引存档，包括1979年至今全球广播，印刷和网络新闻媒体报道的事件。假设一个国家的GDELT事件越多，政治相关性就越大。由于GDELT是几何点的数据集，不能直接知道它属于哪个国家，因此我们需要第二个数据集，即FIPS Codes（联邦信息处理标准出版物代码）的shapefile，可以表示一个国家的边界，两个数据集之间可以JION。（本例进行的前提是两个数据集已经存储好了）

本例中国家只有3000条记录，我们可以广播（Spark Broadcast）这些国家让查询更有效率。

（一）开发任务

• 使用GeoMesa Spark

• 通过地理空间的UDF创造和使用DF。

• 使用多边形计算聚合统计数据。

• 创造一个新的简单特征类型来表示聚合。

• 将结果可视化为等值区域图。

（二）需求背景

GDELT提供了一个全面的时间和位置索引存档，包括1979年至今全球广播，印刷和网络新闻媒体报道的事件。假设一个国家的GDELT事件越多，政治相关性就越大。由于GDELT是几何点的数据集，不能直接知道它属于哪个国家，因此我们需要第二个数据集，即FIPS Codes（联邦信息处理标准出版物代码）的shapefile，可以表示一个国家的边界，两个数据集之间可以JION。（本例进行的前提是两个数据集已经存储好了）

本例中国家只有3000条记录，我们可以广播（Spark Broadcast）这些国家让查询更有效率。

（三）代码分析

/** 
** 当运行Spark Job时，需要运行Jar包。GeoMesa提供了Accumulo、HBase和FS数据存储
**/
$ 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 fipsParams = Map(
  "accumulo.instance.id" -> "instance",
  "accumulo.zookeepers"  -> "zoo1:2181,zoo2:2181,zoo3:2181",
  "accumulo.user"        -> "user",
  "accumulo.password"    -> "password",
  "accumulo.catalog"     -> "fips")
val gdeltParams = Map(
  "accumulo.instance.id" -> "instance",
  "accumulo.zookeepers"  -> "zoo1:2181,zoo2:2181,zoo3:2181",
  "accumulo.user"        -> "user",
  "accumulo.password"    -> "password",
  "accumulo.catalog"     -> "gdelt")
//将数据加载到DF
val fipsDF = spark.read.format("geomesa")
  .options(fipsParams)
  .option("geomesa.feature", "fips")
  .load()
val gdeltDF = spark.read.format("geomesa")
  .options(gdeltParams)
  .option("geomesa.feature", "gdelt")
  .load()
//过滤DF。例如：我们只需要一周内的GDELT数据。
import spark.implicits._
val filteredGdelt = gdeltDF.where("dtg between '2018-01-01 12:00:00' and '2018-01-08 12:00:00'")
//broadcast Join。本例中是将FIPS数据发送到了每个executor。
import org.apache.spark.sql.functions.broadcast
val joinedDF = gdeltDF.join(broadcast(fipsDF), st_contains($"the_geom", $"geom"))
// 聚合
import org.apache.spark.sql.functions.concat
import org.apache.spark.sql.functions.count
import org.apache.spark.sql.functions.first
val aggregateDF = joinedDF.groupBy(concat($"STATEFP", $"COUNTYFP"))
  .agg(first("NAME").as("name"),
       count($"globalEventId").as("eventCount"),
       first("the_geom").as("geom"))
//可视化。首先需要把数据转为GeoJSON格式。
import org.locationtech.geomesa.spark.jts.util.GeoJSONExtensions._
val geojsonDF = aggregateDF.toGeoJSON
//将结果放入内存，收集到driver端并写入文件。或者写入HDFS。
val geoJsonString = geojsonDF.collect.mkString("[",",","]")

//创建一个leaflet map，这个map可以被嵌入web页面，在JS中交互。可视化部分参见：
https://www.geomesa.org/documentation/tutorials/broadcast-join.html