使用Geomesa-spark处理管线数据

ETL部分提供了城市（例如昆明、苏州）的管线数据，管线有不同类型（电力、热水、给水、中水、雨水、交通信号、通信等），某个具体类型的管线数据，也有不同类型（点Point、线Line）。本案例选用昆明市的电力线表，即km_npl。（np代表“电力”，l代表线）

（一）需求描述

数据

ETL部分提供了城市（例如昆明、苏州）的管线数据，管线有不同类型（电力、热水、给水、中水、雨水、交通信号、通信等），某个具体类型的管线数据，也有不同类型（点Point、线Line）。本案例选用昆明市的电力线表，即km_npl。（np代表“电力”，l代表线）

需求

• 根据管线的地理空间位置信息，分析管线之间的安全距离。

• 根据管线的埋深，分析管线垂直方向的安全距离。

（二）主要开发环境与工具

• JDK1.8

• Apache Maven 3.5.2或以上版本

• Kudu集群

• GeoServer 2.12.0

• GeoTools Java API

• Geomesa2.2.0源码包中的geomesa-kudu模块、geomesa-spark-core模块、geomesa-spark-sql模块、geomesa-spark-jts模块

（三）开发步骤

准备数据

ETL提供的数据存储在PostgreSQL，需要将数据从PostgreSQL转移到Kudu中存储。先将PostgreSQL中km_npl表格数据导出为CSV文件。

存储数据

读取CSV文件。

设置Kudu表格的schema，定义simpleFeatureType。km_npl表中geom字段为Geometry类型，具体为十六进制字符串，即WKB（well-known binary）。

import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;

StringBuilder attributes = new StringBuilder();
attributes.append("*geom:Geometry:srid=4326");
SimpleFeatureType sft = 
SimpleFeatureTypes.createType(getTypeName(), attributes.toString());
sft.getUserData().put("geomesa.mixed.geometries", "true");

将WKB传入Geometry对象再进行存储，否则存储失败，表中数据为空。

WKB原数据示例：

0102000020E61000000200000014BF22F2EFA959400ED2E9FF300A39406349C36DF0A95940310B5DA8040A3940

放入Geometry对象后的存储内容示例：

LINESTRING (102.654372 24.95892, 102.65433 24.958804)

具体代码如下：

import com.vividsolutions.jts.geom.Geometry;
import com.vividsolutions.jts.io.WKBReader;
import org.geotools.feature.simple.SimpleFeatureBuilder;

SimpleFeatureBuilder builder = new SimpleFeatureBuilder(getSimpleFeatureType());
byte[] byteArr = new WKBReader().hexToBytes(record.get(46));
Geometry geometry = new WKBReader().read(byteArr);
builder.set("geom",geometry);

GeoServer可视化与查询

在GeoServer数据存储（data store）中选择矢量数据源Kudu(Geomesa)，发布图层并预览。

使用SparkSQL st函数分析Kudu管线数据

读取kudu中存储的管线数据（表格：etl_km_npl），选用st函数分析geom字段，计算管线距离。（需要计算两两之间的管线距离，此处直接展示了笛卡尔积的表格数据，存在数据量多大时的内存加载问题）

val dsParams = Map(
      "kudu.master" -> "host:port",
      "kudu.catalog" -> "tbl")     //kudu table Name
// 创建SparkSession
val sparkSession = SparkSession.builder()
      .appName("etl_km_npl")
      .config("spark.sql.crossJoin.enabled", "true")
      .master("local[*]")
      .getOrCreate()
// 使用 "geomesa" 格式创建DF
val dataFrame = sparkSession.read
      .format("geomesa")
      .options(dsParams)
      .option("geomesa.feature", "tbl")   //simple feature type Name
      .load()
    dataFrame.createOrReplaceTempView("km")
    dataFrame.show(5)
// 查询
/**
* st_distance(Geometry a, Geometry b)   2D笛卡尔距离
* st_distanceSphere(Geometry a, Geometry b)  假设球形地球，近似两个经度/纬度几何之间的最小距离。 * st_distanceSpheroid(Geometry a, Geometry b)  假设WGS84球体，返回两个经度/纬度几何之间的最小距离。
 */
 val sqlQuery = "select k1.gid as gid1,k2.gid as gid2, st_distance (k1.geom,k2.geom) as distance " +"from km k1,km k2 "
 val resultDataFrame = sparkSession.sql(sqlQuery)

部分计算结果如下：

附：遇到的报错及处理

GeoServer发布图层

在GeoServer中新建数据存储（data store）并选择发布图层layer后报错如下：

【解决】报错原因应该是版本冲突。原本的GeoServer版本是2.14.1，换成2.12.0的版本重新安装就没有报错了。

WKB 解析错误

管线数据中有Geometry类型的字段，存储形式为十六进制字符串（例如：0102000020E61000000200000014BF22F2EFA959400ED2E9FF300A39406349C36DF0A95940310B5DA8040A3940），使用字符串的getBytes( )方法，会报如下错误：

【解决】解析方式不对。Geometry十六进制字符串，即WKB（well-known binary），需要使用new WKBReader().hexToBytes(“hexString”)，先将十六进制转为字节数组。