Rename to sparkgeo; initial version of spark_read_geojson

Author: martinstuder

DESCRIPTION

@@ -1,6 +1,6 @@
-Package: sparknyc
+Package: sparkgeo
 Type: Package
-Title: sparknyc: NYC neighborhood matching
+Title: Geospatial Analytics for Spark
 Version: 0.1.0
 Authors@R: c(person(family = "Mirai Solutions GmbH", role = "aut",
                     email = "info@mirai-solutions.com"),
@@ -10,13 +10,13 @@ Authors@R: c(person(family = "Mirai Solutions GmbH", role = "aut",
                     email = "nicola.lambiase@mirai-solutions.com"),
              person("Ömer", "Demirel", role = c("ctb"),
                     email = "omer.demirel@mirai-solutions.com"))
-Description: The 'sparknyc' package is a 'sparklyr' extension package providing NYC geospatial
-  neighborhood matching.
+Description: The 'sparkgeo' package is a 'sparklyr' extension package providing geospatial
+  analytics capabilities to Spark and R.
 Depends: R (>= 3.3.2)
 Imports: sparklyr
 Suggests: dplyr
-License: GPL-3
-SystemRequirements: Java (>= 1.8)
+License: GPL-3 | file LICENSE
+SystemRequirements: Java (>= 1.8), Spark 2.x
 Encoding: UTF-8
 LazyData: true
 RoxygenNote: 6.0.1

NAMESPACE

@@ -1,6 +1,8 @@
 # Generated by roxygen2: do not edit by hand
 
-export(sparknyc_register)
+export(spark_read_geojson)
+export(sparkgeo_register_udfs)
 importFrom(sparklyr,invoke_static)
 importFrom(sparklyr,register_extension)
+importFrom(sparklyr,spark_read_source)
 importFrom(sparklyr,spark_session)

R/dependencies.R

@@ -2,8 +2,8 @@ spark_dependencies <- function(spark_version, scala_version, ...) {
   sparklyr::spark_dependency(
     jars = c(
       system.file(
-        sprintf("java/sparknyc-%s-%s.jar", spark_version, scala_version),
-        package = "sparknyc"
+        sprintf("java/sparkgeo-%s-%s.jar", spark_version, scala_version),
+        package = "sparkgeo"
       )
     ),
     packages = sprintf("harsha2010:magellan:1.0.5-s_%s", scala_version)

R/main.R

@@ -0,0 +1,40 @@
+#' @title Reading GeoJSON files
+#' @description Imports data from GeoJSON files into Spark DataFrames.
+#' @param sc \code{\link[sparklyr]{spark_connection}} provided by sparklyr.
+#' @param name The name to assign to the newly generated table (see also
+#' \code{\link[sparklyr]{spark_read_source}}).
+#' @param path The path to the GeoJSON file. This may be a local path or
+#' an HDFS path.
+#' @param magellanIndex \code{logical} specifying whether geometries should
+#' be indexed when loading the data (see
+#' \url{https://github.com/harsha2010/magellan#creating-indexes-while-loading-data}).
+#' Indexing creates an additional column called "index" which holds the list of
+#' ZOrder curves of the given precision (see argument \code{magellanIndexPrecision}).
+#' Defaults to \code{TRUE}.
+#' @param magellanIndexPrecision \code{integer} specifying the precision to use for creating
+#' the ZOrder curves.
+#' @param ... Additional arguments passed to \code{\link[sparklyr]{spark_read_source}}.
+#' @return A \code{tbl_spark} which provides a \code{dplyr}-compatible reference to a
+#' Spark DataFrame.
+#' @references
+#' \url{https://github.com/harsha2010/magellan}
+#' \url{http://geojson.org/}
+#' @family Spark serialization routines
+#' @seealso \code{\link[sparklyr]{spark_read_source}}
+#' @keywords file, connection
+#' @importFrom sparklyr spark_read_source
+#' @export
+spark_read_geojson <- function(sc, name, path, magellanIndex = TRUE, magellanIndexPrecision = 30L, ...) {
+  spark_read_source(
+    sc = sc,
+    name = name,
+    source = "magellan",
+    options = list(
+      "type" = "geojson",
+      "magellan.index" = ifelse(magellanIndex, "true", "false"),
+      "magellan.index.precision" = as.character(magellanIndexPrecision),
+      "path" = path
+    ),
+    ...
+  )
+}

R/spark_read_geojson.R

@@ -0,0 +1,10 @@
+#' @title Register Spark User-Defined Functions (UDFs)
+#' @description Registers UDFs with Spark
+#' @param sc \code{\link[sparklyr]{spark_connection}} provided by sparklyr.
+#' @importFrom sparklyr invoke_static
+#' @importFrom sparklyr spark_session
+#' @export
+sparkgeo_register_udfs <- function(sc) {
+  sparklyr::invoke_static(sc, "com.miraisolutions.spark.geo.UDF", "register", spark_session(sc))
+  invisible()
+}

R/sparkgeo_register_udfs.R

@@ -1,37 +1,17 @@
-package com.miraisolutions.spark.nyc
+package com.miraisolutions.spark.geo
 
-import org.apache.spark.sql.{DataFrame, SparkSession}
-import org.apache.spark.sql.magellan.dsl.expressions._
-import org.apache.spark.sql.functions._
+import org.apache.spark.sql.SparkSession
+import magellan.{Point, Polygon}
+import scala.collection.immutable.Map
 
-object Main {
+object UDF {
+  def register(spark: SparkSession): Unit = {
+    // See https://github.com/harsha2010/magellan#spatial-joins
+    magellan.Utils.injectRules(spark)
 
-  private def read_neighborhoods(spark: SparkSession, path: String): DataFrame = {
-    import spark.implicits._
-
-    spark.read
-      .format("magellan")
-      .option("type", "geojson")
-      .option("magellan.index", "true")
-      .load(path)
-      .select($"polygon", $"metadata"("neighborhood").as("name"))
-      .cache()
-  }
-
-  private var neighborhoods: DataFrame = _
-
-  private val match_neighborhood = (latitude: Double, longitude: Double) => {
-    neighborhoods
-      .filter(point(lit(longitude), lit(latitude)) within col("polygon"))
-      .select("name")
-      .collect()
-      .map(_.getString(0))
-      .headOption
-      .getOrElse(null)
-  }
-
-  def register_nyc(spark: SparkSession, neighborhoodFile: String): Unit = {
-    neighborhoods = read_neighborhoods(spark, neighborhoodFile)
-    spark.udf.register("neighborhood", match_neighborhood)
+    spark.udf.register("point", (latitude: Double, longitude: Double) => Point(longitude, latitude))
+    spark.udf.register("within", (point: Point, polygon: Polygon) => point.within(polygon))
+    spark.udf.register("metadata_string", (metadata: Map[String, Any], name: String) =>
+      metadata(name).asInstanceOf[String])
   }
 }