Geomesa + SparkSQL-Integrationsproblem

Question

Mein Setup ist ein 3-Knoten-Cluster, der in AWS ausgeführt wird. Ich habe bereits meine Daten (30 Millon Zeilen) aufgenommen und habe keine Probleme beim Ausführen von Abfragen mit Jupyter Notebook. Aber jetzt versuche ich eine Abfrage mit Funke und Java auszuführen, wie im folgenden Ausschnitt zu sehen ist.

public class SparkSqlTest { 

    private static final Logger log = Logger.getLogger(SparkSqlTest.class); 


    public static void main(String[] args) { 
     Map<String, String> dsParams = new HashMap<>(); 
     dsParams.put("instanceId", "gis"); 
     dsParams.put("zookeepers", "server ip"); 
     dsParams.put("user", "root"); 
     dsParams.put("password", "secret"); 
     dsParams.put("tableName", "posiciones"); 

     try { 
      DataStoreFinder.getDataStore(dsParams); 
      SparkConf conf = new SparkConf(); 
      conf.setAppName("testSpark"); 
      conf.setMaster("yarn"); 
      SparkContext sc = SparkContext.getOrCreate(conf); 
      SparkSession ss = SparkSession.builder().config(conf).getOrCreate(); 

      Dataset<Row> df = ss.read() 
       .format("geomesa") 
       .options(dsParams) 
       .option("geomesa.feature", "posicion") 
       .load(); 
      df.createOrReplaceTempView("posiciones"); 

      long t1 = System.currentTimeMillis(); 
      Dataset<Row> rows = ss.sql("select count(*) from posiciones where id_equipo = 148 and fecha_hora >= '2015-04-01' and fecha_hora <= '2015-04-30'"); 
      long t2 = System.currentTimeMillis(); 
      rows.show(); 

      log.info("Tiempo de la consulta: " + ((t2 - t1)/1000) + " segundos."); 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } 
    } 
} 

ich den Code in meinem Master-EC2-Box laden (im jupyter notebook image), und es laufen die folgenden Befehle:

docker cp myjar-0.1.0.jar jupyter:myjar-0.1.0.jar 
docker exec jupyter sh -c '$SPARK_HOME/bin/spark-submit --master yarn --class mypackage.SparkSqlTest file:///myjar-0.1.0.jar --jars $GEOMESA_SPARK_JARS' 

Aber ich habe folgende Fehlermeldung:

17/09/15 19:45:01 INFO HSQLDB4AD417742A.ENGINE: dataFileCache open start 
17/09/15 19:45:02 INFO execution.SparkSqlParser: Parsing command: posiciones 
17/09/15 19:45:02 INFO execution.SparkSqlParser: Parsing command: select count(*) from posiciones where id_equipo = 148 and fecha_hora >= '2015-04-01' and fecha_hora <= '2015-04-30' 
java.lang.RuntimeException: Could not find a SpatialRDDProvider 
at org.locationtech.geomesa.spark.GeoMesaSpark$$anonfun$apply$2.apply(GeoMesaSpark.scala:33) 
at org.locationtech.geomesa.spark.GeoMesaSpark$$anonfun$apply$2.apply(GeoMesaSpark.scala:33) 

Irgendwelche Ideen, warum das passiert?

Answer 1

ich sortiert endlich, mein Problem war, dass ich nicht die folgenden Einträge in meinem pom.xml

<dependency> 
     <groupId>org.locationtech.geomesa</groupId> 
     <artifactId>geomesa-accumulo-spark_2.11</artifactId> 
     <version>${geomesa.version}</version> 
    </dependency> 

    <dependency> 
     <groupId>org.locationtech.geomesa</groupId> 
     <artifactId>geomesa-spark-converter_2.11</artifactId> 
     <version>${geomesa.version}</version> 
    </dependency>