[geomesa-users] Geomesa Spark Java API

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index] [List Home]

[geomesa-users] Geomesa Spark Java API

From: Jose Bujalance <joseab56@xxxxxxxxx>
Date: Thu, 16 Mar 2017 15:10:36 +0100
Delivered-to: geomesa-users@xxxxxxxxxxxxxxxx
List-archive: <https://dev.locationtech.org/mhonarc/lists/geomesa-users>
List-help: <mailto:geomesa-users-request@locationtech.org?subject=help>
List-subscribe: <https://dev.locationtech.org/mailman/listinfo/geomesa-users>, <mailto:geomesa-users-request@locationtech.org?subject=subscribe>
List-unsubscribe: <https://dev.locationtech.org/mailman/options/geomesa-users>, <mailto:geomesa-users-request@locationtech.org?subject=unsubscribe>

Hi again,

I am trying the new Java API for Geomesa-Spark provided in the 1.3.1 version, but I am having some troubles.

First of all, I have tested that everything works fine querying my Accumulo datastore through the Spark shell using the geomesa-accumulo-spark-runtime_2.11-1.3.1.jar. This is how my scala code looks like:

import org.apache.hadoop.conf.Configuration

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

import org.locationtech.geomesa.spark._

import org.geotools.data.{DataStoreFinder, Query}

import org.geotools.factory.CommonFactoryFinder

import org.geotools.filter.text.ecql.ECQL

import scala.collection.JavaConversions._

// Accumulo datastore params

val params = Map(

"instanceId" -> "hdp-accumulo-instance",

"zookeepers" -> "hdf-sb-a.praxedo.net:2181,hdf-sb-b.praxedo.net:2181",

"user" -> "root",

"password" -> "praxedo",

"tableName" -> "Geoloc_Praxedo"

)

// set the configuration to the existant SparkContext

val conf = sc.getConf

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

conf.set("spark.kryo.registrator", classOf[GeoMesaSparkKryoRegistrator].getName)

val sc = SparkContext.getOrCreate(conf)

// create RDD with a geospatial query using Geomesa functions

val spatialRDDProvider = GeoMesaSpark(params)

val filter = ECQL.toFilter("BBOX(coords, 2.249294, 48.815215, 2.419337, 48.904295)")

val query = new Query("history_1M", filter)

val resultRDD = spatialRDDProvider.rdd(new Configuration, sc, params, query)

resultRDD.count

This code works fine, giving the expected result.

Now I am trying to do the same thing on Java. This is how my code looks like:

package com.praxedo.geomesa.geomesa_spark;

import org.apache.hadoop.conf.Configuration;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaSparkContext;

import org.geotools.data.Query;

import org.geotools.filter.text.cql2.CQLException;

import org.geotools.filter.text.ecql.ECQL;

import org.locationtech.geomesa.spark.api.java.*;

import java.io.IOException;

import java.util.HashMap;

import java.util.Map;

public class Test {

private final static String ACCUMULO_INSTANCE = "hdp-accumulo-instance";

private final static String ACCUMULO_ZOOKEEPERS = "hdf-sb-a.praxedo.net:2181,hdf-sb-b.praxedo.net:2181";

private final static String ACCUMULO_USER = "root";

private final static String ACCUMULO_PASSWORD = "password";

private final static String GEOMESA_CATALOG = "Geoloc_Praxedo";

private final static String GEOMESA_FEATURE = "history_1M";

public static void main(String[] args) throws IOException, CQLException {

//Spark configuration

SparkConf conf = new SparkConf().setAppName("MyAppName").setMaster("local[*]");

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

conf.set("spark.kryo.registrator", "org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator");

JavaSparkContext jsc = new JavaSparkContext(conf);

//Datastore configuration

Map<String, String> parameters = new HashMap<>();

parameters.put("instanceId", ACCUMULO_INSTANCE);

parameters.put("zookeepers", ACCUMULO_ZOOKEEPERS);

parameters.put("user", ACCUMULO_USER);

parameters.put("password", ACCUMULO_PASSWORD);

parameters.put("tableName", GEOMESA_CATALOG);

JavaSpatialRDDProvider provider = JavaGeoMesaSpark.apply(parameters);

String predicate = "BBOX(coords, 2.249294, 48.815215, 2.419337, 48.904295)";

Query query = new Query(GEOMESA_FEATURE, ECQL.toFilter(predicate));

JavaSpatialRDD resultRDD = provider.rdd(new Configuration(), jsc, parameters, query);

System.out.println("Number of RDDs: " + resultRDD.count());

System.out.println("First RDD: " + resultRDD.first());

}

And here are the dependencies I am importing with Maven:

<dependency>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>3.8.1</version>

<scope>test</scope>

</dependency>

<dependency>

<groupId>org.apache.accumulo</groupId>

<artifactId>accumulo-core</artifactId>

<version>1.7.0</version>

</dependency>

<dependency>

<groupId>org.apache.accumulo</groupId>

<artifactId>accumulo-fate</artifactId>

<version>1.7.0</version>

</dependency>

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<version>2.0.0</version>

</dependency>

<dependency>