[geomesa-dev] [JIRA] (GEOMESA-2794) Docs - add example for reading/writi

["Emilio Lahr-Vivaz (JIRA)" <jira@xxxxxxxxxxxxxxxxxxxxx>]

Title: Message Title

Emilio Lahr-Vivaz created an issue

GeoMesa / GEOMESA-2794

Docs - add example for reading/writing to postgis in spark

Issue Type:	Improvement
Assignee:	Unassigned
Created:	04/Mar/20 9:40 AM
Priority:	Major
Reporter:	Emilio Lahr-Vivaz

From Jeff Gold:

STEPS:

1. You have to add the kryo serializers to the spark submit and the geomesa geotools jar (geomesa-gt-spark-runtime_2.11-2.4.0.jar). Here are the details that we used:

    export SPARK_SUBMIT_OPTIONS="--jars /opt/depZep/geomesa-accumulo-spark-runtime_2.11-2.4.0.jar,/opt/depZep/geomesa-gt-spark-runtime_2.11-2.4.0.jar,/opt/depZep/geomesa-cqengine-datastore_2.11-2.4.0.jar,/opt/depZep/jai_core-1.1.3.jar --conf spark.dynamicAllocation.enabled=true --conf spark.shuffle.service.enabled=true --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.kryo.registrator=org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator"

2. The other tricky thing was setting the table up correctly with primary key as string, etc. You also need to grant read/write to the user you connect with in Spark
3. Here is reading from the existing table (in Scala)

import org.apache.spark.sql.SparkSession

import org.locationtech.geomesa.spark.GeoMesaSparkKryoRegistrator

import org.locationtech.geomesa.spark.jts._

 

val spark: SparkSession = SparkSession.builder()

    .appName("testSpark")

    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

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

    .master("yarn-client")

    .getOrCreate()

    .withJTS

   

val PGparams = Map(

  "geotools" -> "true",

  "dbtype"   -> "postgis",

  "host"     -> "MACHINE1",

  "user"     -> "USER",

  "passwd"   -> "PASSWORD",

  "port"     -> "5432",

  "database" -> "gistest")

 

val gisDF = spark.read.format("geomesa")

  .format("geomesa")

  .options(PGparams)

  .option("geomesa.feature", "PG_TABLE_NAME")

  .load()

 

 

4. Writing new data I pulled from GeoMesa in to this table.. sample record.

import org.apache.spark.sql.types.IntegerType

 

val r = scala.util.Random

 

// Pull AIS sample data in this case

val GMparams = Map(

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

    "zookeepers" -> "MACHINE1",

    "user"       -> "root",

    "password"   -> "PASSWORD",

    "tableName"  -> "geomesa220.AISFEATURE")

   

    // Create SparkSession

val sparkSession = SparkSession.builder().appName("GeoMesa test").config("spark.sql.crossJoin.enabled", "true").getOrCreate()

 

// Create DataFrame using the "geomesa" format

val df = sparkSession.read.format("geomesa").options(GMparams).option("geomesa.feature", "AIS").load()

// Match record schema to POSTGIS table schema

val df2 = df.withColumnRenamed("_fid_", "name").select("name", "geom").limit(5)

 

// See sample

df2.show(100)

 

 

import org.locationtech.geomesa.spark.GeoMesaSpark

df2.write.format("geomesa")

  .options(PGparams)

  .option("geomesa.feature", " PG_TABLE_NAME").save("geomesa")

Add Comment

Get Jira notifications on your phone! Download the Jira Cloud app for Android or iOS
This message was sent by Atlassian Jira (v1001.0.0-SNAPSHOT#100121-sha1:d2b7993)