import java.text.SimpleDateFormat
import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.geotools.data.{DataStoreFinder, Query}
import org.geotools.factory.CommonFactoryFinder
import org.locationtech.geomesa.accumulo.data._
import org.locationtech.geomesa.spark.GeoMesaSpark
import org.opengis.filter.Filter

import scala.collection.JavaConversions._

name := "GeoMesaSparkExample"

version := "0.1"

scalaVersion := "2.11.8"

// Add necessary resolvers
resolvers ++= Seq(
  "locationtech-releases" at "https://repo.locationtech.org/content/groups/releases",
  "boundlessgeo" at "https://repo.boundlessgeo.com/main",
  "osgeo" at "http://download.osgeo.org/webdav/geotools",
  "conjars.org" at "http://conjars.org/repo",
  "media.javax" at "http://maven.geotoolkit.org"
)

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.0.2" % "provided"
  , "org.apache.spark" %% "spark-catalyst" % "2.0.2" % "provided"
  , "org.apache.spark" %% "spark-sql" % "2.0.2" % "provided"
/**  , "org.apache.spark" %% "spark-yarn" % "2.0.2" % "provided" **/
  , "org.locationtech.geomesa" %% "geomesa-accumulo-datastore" % "1.3.1"
  , "org.locationtech.geomesa" %% "geomesa-accumulo-spark" % "1.3.1"
  , "org.locationtech.geomesa" %% "geomesa-spark-sql" % "1.3.1"
  , "ch.qos.logback" % "logback-classic" % "1.1.7"
  , "com.typesafe.scala-logging" %% "scala-logging" % "3.5.0"
  , "org.apache.accumulo" % "accumulo-core" % "1.7.3"
)

package examples

import java.text.SimpleDateFormat

import org.apache.hadoop.conf.Configuration
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.geotools.data.{DataStoreFinder, Query}
import org.geotools.factory.CommonFactoryFinder
import org.geotools.filter.text.ecql.ECQL
import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
import org.locationtech.geomesa.spark.GeoMesaSpark
import org.opengis.feature.simple.SimpleFeature

import scala.collection.JavaConversions._

object CountByDay {

  val params = Map(
    "instanceId" -> "BISSOL_CONSULTING",
    // zookeeper info can be found in ACCUMULO_HOME/conf/accumulo-site.xml
    "zookeepers" -> "localhost:2181",
    "user"       -> "root",
    "password"   -> "password",
    // no authentication required for local dev env setup
    //"auths"      -> "USER,ADMIN",
    "tableName"  -> "myNamespace.gdelt"
    )

  // see geomesa-tools/conf/sfts/gdelt/reference.conf
  val typeName = "gdelt"
  val geom     = "geom"
  val date     = "dtg"

  val bbox   = "-80, 35, -79, 36"
  val during = "2014-01-01T00:00:00.000Z/2014-01-31T12:00:00.000Z"

  // val filter = s"bbox($geom, $bbox) AND $date during $during"
  val filter = s"bbox($geom, $bbox)"

  def main(args: Array[String]) {
    // Get a handle to the data store
    val ds = DataStoreFinder.getDataStore(params).asInstanceOf[AccumuloDataStore]

    // Construct a CQL query to filter by bounding box
    val q = new Query(typeName, ECQL.toFilter(filter))

    // Configure Spark
    val conf = new SparkConf().setAppName("testSpark")
    val sc = SparkContext.getOrCreate(conf)

    // Get the appropriate spatial RDD provider
    val spatialRDDProvider = GeoMesaSpark(params)

    // Get an RDD[SimpleFeature] from the spatial RDD provider
    val rdd = spatialRDDProvider.rdd(new Configuration, sc, params, q)

    // Collect the results and print
    countByDay(rdd).collect().foreach(println)
    println("\n")

    ds.dispose()
  }

  // extract the SQLDATE from each SimpleFeature and truncating it to the day resolution
  def countByDay(rdd: RDD[SimpleFeature], dateField: String = "dtg") = {
    val dayAndFeature = rdd.mapPartitions { iter =>
      val df = new SimpleDateFormat("yyyyMMdd")
      val ff = CommonFactoryFinder.getFilterFactory2
      val exp = ff.property(dateField)
      iter.map { f => (df.format(exp.evaluate(f).asInstanceOf[java.util.Date]), f) }
    }
    dayAndFeature.map( x => (x._1, 1)).reduceByKey(_ + _)
  }
}

import org.geotools.data.{DataStoreFinder, Query}
import org.geotools.factory.CommonFactoryFinder
import org.geotools.filter.text.ecql.ECQL
//import org.opengis.filter.Filter
import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
import org.locationtech.geomesa.spark.GeoMesaSpark
// org.locationtech.geomesa.compute.spark.GeoMesaSpark
// seems to be deprecated, see https://github.com/locationtech/geomesa/blob/master/geomesa-accumulo/geomesa-accumulo-compute/src/main/scala/org/locationtech/geomesa/compute/spark/GeoMesaSpark.scala

import org.opengis.feature.simple.SimpleFeature

import java.text.SimpleDateFormat
import scala.collection.JavaConversions._
import org.apache.hadoop.conf.Configuration

val params = Map(
  "instanceId" -> "BISSOL_CONSULTING",
  // zookeeper info can be found in ACCUMULO_HOME/conf/accumulo-site.xml
  "zookeepers" -> "localhost:2181",
  "user"       -> "root",
  "password"   -> "password",
  // no authentication required for local dev env setup
  //"auths"      -> "USER,ADMIN",
  "tableName"  -> "myNamespace.gdelt"
  )

val featureName = "gdelt"
val geom     = "geom"
val date     = "dtg"

val bbox   = "-80, 35, -79, 36"
val during = "2017-01-05T00:00:00.000Z/2017-01-06T12:00:00.000Z"

// val filter = s"bbox($geom, $bbox) AND $date during $during"
val filter = s"bbox($geom, $bbox)"
// Get a handle to the data store
val ds = DataStoreFinder.getDataStore(params).asInstanceOf[AccumuloDataStore]

// Construct a CQL query to filter by bounding box
val q = new Query(featureName, ECQL.toFilter(filter))

// Get the appropriate spatial RDD provider
val spatialRDDProvider = GeoMesaSpark(params)

// Get an RDD[SimpleFeature] from the spatial RDD provider
// this one requires import org.apache.hadoop.conf.Configuration
// THIS EXTRACTS THE KEYS ONLY
val rdd = spatialRDDProvider.rdd(new Configuration, sc, params, q)

// IF IT GETS STUCK HERE: either connection details are wrong
// or you are not using the dedicated Accumulo namespace
// which has all the geomesa dependencies.

rdd.take(2)
/** 
result should look similar to this one:
res0: Array[org.opengis.feature.simple.SimpleFeature] = Array(ScalaSimpleFeature:31a70f3f6e748374b22f24e0094e76b4, ScalaSimpleFeature:51117014214fa48a009c33c37e2784d7)

you can find these keys via Accumulo as well, e.g.:
scan -t myNamespace.gdelt_gdelt_records_v2 -r 51117014214fa48a009c33c37e2784d7
**/

// extract the date from each SimpleFeature and truncating it to the day resolution
val dayAndFeature = rdd.mapPartitions { iter =>
  val df = new SimpleDateFormat("yyyyMMdd")
  val ff = CommonFactoryFinder.getFilterFactory2
  // extract date, "dtg" is the name of the date field
  val exp = ff.property("dtg")
  iter.map { f => (df.format(exp.evaluate(f).asInstanceOf[java.util.Date]), f) }
}

// group by the day and count up the number of events in each group.

val countByDay = dayAndFeature.map( x => (x._1, 1)).reduceByKey(_ + _)

countByDay.collect().foreach(println)

package examples

import java.text.SimpleDateFormat

import org.apache.hadoop.conf.Configuration
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.geotools.data.{DataStoreFinder, Query}
import org.geotools.factory.CommonFactoryFinder
import org.geotools.filter.text.ecql.ECQL
import org.locationtech.geomesa
import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
import org.locationtech.geomesa.spark.GeoMesaSpark
import com.vividsolutions.jts.geom.Point
import org.opengis.feature.simple.SimpleFeature

import scala.collection.JavaConversions._

object SpatialDensities {

  val params = Map(
    "instanceId" -> "BISSOL_CONSULTING",
    // zookeeper info can be found in ACCUMULO_HOME/conf/accumulo-site.xml
    "zookeepers" -> "localhost:2181",
    "user"       -> "root",
    "password"   -> "password",
    // no authentication required for local dev env setup
    //"auths"      -> "USER,ADMIN",
    "tableName"  -> "myNamespace.gdelt"
    )


  val typeName = "gdelt"
  val geom     = "geom"
  val bbox   = "-180, -90, 180, 90"
  val cartoClass = "EPSG:4326"

  //val f = bbox("geom", -180, -90, 180, 90, "EPSG:4326")
  //  val filter = s"""bbox($geom, $bbox, $cartoClass)"""
  val filter = s"bbox($geom, $bbox)"

  def main(args: Array[String]) {
  
    // Get a handle to the data store
    val ds = DataStoreFinder.getDataStore(params).asInstanceOf[AccumuloDataStore]

    // Construct new query
    //val q = new Query("gdelt", f)
    val q = new Query(typeName, ECQL.toFilter(filter))
    
    // Configure Spark
    val conf = new SparkConf().setAppName("testSpark")
    val sc = SparkContext.getOrCreate(conf)

    // Get the appropriate spatial RDD provider
    val spatialRDDProvider = GeoMesaSpark(params)

    val queryRDD = spatialRDDProvider.rdd(new Configuration, sc, params, q)

    // Project (in the relational sense) the SimpleFeature to a 2-tuple of (GeoHash, 1)
    val discretized = queryRDD.map {
      f => (
        geomesa.utils.geohash.GeoHash(f.getDefaultGeometry.asInstanceOf[Point], 25)
        , 1
      )
    }

    // group by grid cell and count the number of features per cell.
    val density = discretized.reduceByKey(_ + _)

    density.collect.foreach(println)

    ds.dispose()
  }
}

import java.text.SimpleDateFormat

import org.apache.hadoop.conf.Configuration
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
import org.geotools.data.{DataStoreFinder, Query}
import org.geotools.factory.CommonFactoryFinder
import org.geotools.filter.text.ecql.ECQL
import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
import org.locationtech.geomesa.spark.GeoMesaSpark
import org.opengis.feature.simple.SimpleFeature

import scala.collection.JavaConversions._

object SparkSQLExample {

  val params = Map(
    "instanceId" -> "BISSOL_CONSULTING",
    // zookeeper info can be found in ACCUMULO_HOME/conf/accumulo-site.xml
    "zookeepers" -> "localhost:2181",
    "user"       -> "root",
    "password"   -> "password",
    // no authentication required for local dev env setup
    //"auths"      -> "USER,ADMIN",
    "tableName"  -> "myNamespace.gdelt"
    )

  def main(args: Array[String]) {

    // Create SparkSession
    val sparkSession = SparkSession.builder()
      .appName("testSpark")
      .config("spark.sql.crossJoin.enabled", "true")
      .master("local[*]")
      .getOrCreate()

    // Create DataFrame using the "geomesa" format
    val dataFrame = sparkSession
      .read
      // register the GeoMesa SparkSQL data source
      .format("geomesa")
      .options(params)
      // tell GeoMesa to use the feature type named gdelt
      .option("geomesa.feature", "gdelt") /** typeName from prev exercise **/
      .load()

    dataFrame.createOrReplaceTempView("gdelt")

    // Query against the "chicago" schema
    val sqlQuery = "select * from gdelt where st_contains(st_makeBBOX(-80.0, 35.0, -79.0, 36.0), geom)"
    val resultDataFrame = sparkSession.sql(sqlQuery)

    resultDataFrame.show
  }
}