前言

本篇笔记是关于GeoMesaSpark的探索，使用语言为Java，实现的功能为：

• 向hbase中插入数据
• 从表中读取满足条件的数据

XML配置文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.fr</groupId><artifactId>geomesaSpark</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><name>geomesaSpark</name><url>http://maven.apache.org</url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><gt.version>20.0</gt.version><geomesa.version>2.2.0</geomesa.version><hadoop.version>2.6.4</hadoop.version><hbase.version>1.3.5</hbase.version><spark.version>2.3.1</spark.version><scala.version>2.11.7</scala.version></properties><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency><!-- geomesa --><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-spark-core_2.11</artifactId><version>${geomesa.version}</version></dependency><!-- geotools --><dependency><groupId>org.geotools</groupId><artifactId>gt-data</artifactId><version>${gt.version}</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-grid</artifactId><version>23.0</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-cql</artifactId><version>23.0</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-geojson</artifactId><version>23.0</version></dependency><!-- spark --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>${spark.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.2.0</version><scope>provided</scope></dependency></dependencies><repositories><repository><id>osgeo</id><name>Open Source Geospatial Foundation Repository</name><url>http://download.osgeo.org/webdav/geotools/</url></repository></repositories>
</project>

存储数据代码

从本地读取一个文件，将数据存储到hbase库中

package cn.fr.geomesaSpark;import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.rdd.RDD;
import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.factory.Hints;
import org.geotools.feature.simple.SimpleFeatureBuilder;
import org.locationtech.geomesa.spark.GeoMesaSpark;
import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;import scala.Tuple2;public class SparkSave {/*** 将java.util.Map转化为scala.collection.immutable.Map* @author yummy* @date 2020年5月18日 上午10:51:08* @param javaMap* @return*/public static scala.collection.immutable.Map<String, String> toScalaImmutableMap(java.util.Map<String, String> javaMap) {List<scala.Tuple2<String, String>> list= new ArrayList();for (final java.util.Map.Entry<String, String> entry : javaMap.entrySet()) {list.add(scala.Tuple2.apply(entry.getKey(), entry.getValue()));}final scala.collection.Seq<Tuple2<String, String>> seq = scala.collection.JavaConverters.asScalaBufferConverter(list).asScala().toSeq();return (scala.collection.immutable.Map<String, String>) scala.collection.immutable.Map$.MODULE$.apply(seq);}public static  Map getParams() {Map params = new HashMap();params.put("hbase.zookeepers", "localhost");params.put("hbase.catalog", "demo2");return params;}public static void main(String[] args) throws IOException {//获取Features SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("testSpark");sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");sparkConf.set("spark.kryo.registrator", "org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator");sparkConf.set("geotools", "true");SparkContext sc = SparkContext.getOrCreate(sparkConf);JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);//文件存储路径JavaRDD<String> brRDD = jsc.textFile("file:///Users/geoMesaTest.txt");JavaRDD<SimpleFeature> simpleFeatureRDD = brRDD.map(new Function<String, SimpleFeature>() {public SimpleFeature call(String line) throws Exception {String[] split = line.split(",");//创建datastoreDataStore datastore = DataStoreFinder.getDataStore(getParams());// 创建schema SimpleFeatureType sft = SimpleFeatureTypes.createType("demo2", "taxiId:String,dtg:Date,*geom:Point:srid=4326,description:String");  //sft.getUserData().put(SimpleFeatureTypes.DEFAULT_DATE_KEY, "dtg"); //设置xz3索引时间间隔为天sft.getUserData().put("geomesa.xz3.interval", "day");//设置时空索引时间字段为datesft.getUserData().put("geomesa.index.dtg", "dateAttr");//设置索引精度sft.getUserData().put("geomesa.xz.precision", 10);datastore.createSchema(sft); //获取FeatureSimpleFeatureBuilder builder = new SimpleFeatureBuilder(sft);  DateTimeFormatter dateFormat = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss", Locale.US);  builder.set("taxiId", split[0]);  builder.set("dtg", Date.from(LocalDateTime.parse(split[1], dateFormat).toInstant(ZoneOffset.UTC)));  builder.set("geom", split[2]);  builder.set("description", split[3]);  builder.featureUserData(Hints.USE_PROVIDED_FID, Boolean.TRUE);  return builder.buildFeature(split[0]);}});RDD<SimpleFeature> rdd = simpleFeatureRDD.rdd();GeoMesaSpark.apply(getParams()).save(rdd, toScalaImmutableMap(getParams()), "demo2");}}

Users/geoMesaTest.txt 文件的内容为：

京P 12377,2008-02-02 13:30:49,POINT (116.31412 39.89454),这是一辆套牌车
冀A 11386,2008-02-10 15:30:49,POINT (118.31412 42.89454),这是一辆正常车

查询数据代码

package cn.fr.geomesaSpark;
/*** Spark查询语句*/
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;import org.apache.hadoop.conf.Configuration;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.data.Query;
import org.geotools.filter.text.cql2.CQL;
import org.geotools.filter.text.cql2.CQLException;
import org.geotools.filter.text.ecql.ECQL;
import org.locationtech.geomesa.spark.GeoMesaSpark;
import org.locationtech.geomesa.spark.SpatialRDD;
import org.locationtech.geomesa.spark.SpatialRDDProvider;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.filter.Filter;import scala.Function1;
import scala.Tuple2;
import scala.runtime.BoxedUnit;public class geomesaSparkDemo {/*** 将java.util.Map转化为scala.collection.immutable.Map* @author yummy* @date 2020年5月18日 上午10:51:08* @param javaMap* @return*/public static scala.collection.immutable.Map<String, String> toScalaImmutableMap(java.util.Map<String, String> javaMap) {List<scala.Tuple2<String, String>> list= new ArrayList();//final java.util.List<scala.Tuple2<String, String>> list = new ArrayList<>(javaMap.size());for (final java.util.Map.Entry<String, String> entry : javaMap.entrySet()) {list.add(scala.Tuple2.apply(entry.getKey(), entry.getValue()));}final scala.collection.Seq<Tuple2<String, String>> seq = scala.collection.JavaConverters.asScalaBufferConverter(list).asScala().toSeq();return (scala.collection.immutable.Map<String, String>) scala.collection.immutable.Map$.MODULE$.apply(seq);}public static void main(String[] args) throws IOException, CQLException {SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("testSpark");sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");sparkConf.set("spark.kryo.registrator", "org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator");sparkConf.set("geotools", "true");SparkContext sc = SparkContext.getOrCreate(sparkConf);JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);HashMap params = new HashMap();params.put("hbase.catalog", "demo2");  params.put("hbase.zookeepers", "localhost");DataStore datastore = DataStoreFinder.getDataStore(params);String typeName = datastore.getTypeNames()[0];Filter filter = CQL.toFilter("taxiId='京P 12377'");//String during = "dtg DURING 2159-01-01T00:00:00.000Z/2169-01-01T00:00:00.000Z";  //Filter filter = ECQL.toFilter(during);//设置过滤条件Query query = new Query(typeName,filter);//不设置过滤条件//Query query = new Query(typeName);SpatialRDD rdd = GeoMesaSpark.apply(params).rdd(new Configuration(), sc, toScalaImmutableMap(params), query); rdd.toJavaRDD().foreach(new VoidFunction<SimpleFeature>() {public void call(SimpleFeature t) throws Exception {// TODO Auto-generated method stubSystem.out.println(t);}});}}

备注

本人在下载添加需要的jar包时遇到依赖jar包下载不成功的情况，pom文件报错，提示缺少相应jar包，以下是我下载解决pom文件报错时添加的所有jar包。如果添加上边的xml文件pom文件可以使用的话，这部分可以忽略。
完整的xml文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.fr</groupId><artifactId>geomesaSpark</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><name>geomesaSpark</name><url>http://maven.apache.org</url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><gt.version>20.0</gt.version><geomesa.version>2.2.0</geomesa.version><hadoop.version>2.6.4</hadoop.version><hbase.version>1.3.5</hbase.version><spark.version>2.3.1</spark.version><scala.version>2.11.7</scala.version></properties><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency><!-- geomesa --><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-hbase-datastore_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-spark-core_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-feature-all_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-hbase-jobs_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-hbase-spark_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-fixedwidth_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-jdbc_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-json_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-xml_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-avro_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-common_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-scripting_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-text_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-all_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-convert-redis-cache_2.11</artifactId><version>${geomesa.version}</version></dependency><dependency><groupId>org.locationtech.geomesa</groupId><artifactId>geomesa-jobs_2.11</artifactId><version>${geomesa.version}</version></dependency><!-- geotools --><dependency><groupId>org.geotools</groupId><artifactId>gt-data</artifactId><version>${gt.version}</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-grid</artifactId><version>23.0</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-cql</artifactId><version>23.0</version></dependency><dependency><groupId>org.geotools</groupId><artifactId>gt-geojson</artifactId><version>23.0</version></dependency><!-- spark --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>${spark.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.2.0</version><scope>provided</scope></dependency><!-- hadoop --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop.version}</version></dependency><!-- hbase --><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-protocol</artifactId><version>${hbase.version}</version><scope>runtime</scope></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>${hbase.version}</version><scope>runtime</scope></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-common</artifactId><version>${hbase.version}</version><scope>runtime</scope></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>${hbase.version}</version><scope>runtime</scope></dependency><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-annotations</artifactId><version>${hbase.version}</version><scope>runtime</scope></dependency><!-- specs2 --><dependency><groupId>org.specs2</groupId><artifactId>specs2-core_2.11</artifactId><version>4.3.2</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-junit_2.11</artifactId><version>4.3.2</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-common_2.11</artifactId><version>4.3.2</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-fp_2.11</artifactId><version>4.3.2</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-matcher_2.11</artifactId><version>4.3.2</version><scope>test</scope></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-compiler</artifactId><version>${scala.version}</version><scope>provided</scope></dependency><dependency><groupId>org.scala-sbt</groupId><artifactId>test-interface</artifactId><version>1.0</version></dependency><dependency><groupId>net.sf.saxon</groupId><artifactId>Saxon-HE</artifactId><version>9.7.0-20</version></dependency><dependency><groupId>com.github.pureconfig</groupId><artifactId>pureconfig_2.11</artifactId><version>0.9.1</version></dependency><dependency><groupId>com.github.pureconfig</groupId><artifactId>pureconfig-macros_2.11</artifactId><version>0.9.1</version></dependency></dependencies><repositories><repository><id>osgeo</id><name>Open Source Geospatial Foundation Repository</name><url>http://download.osgeo.org/webdav/geotools/</url></repository></repositories>
</project>

写在最后

本人系工作需要，接触GeoMesa的小白童鞋，若笔记中有问题的地方，麻烦大佬指正哦~