Reorganize to include examples

Author: jpolchlo

build.sbt

@@ -147,11 +147,22 @@ val vpExtraSettings = Seq(
 // micrositeBaseUrl := "/vectorpipe"
 // micrositeDocumentationUrl := "/vectorpipe/latest/api/#vectorpipe.package" /* Location of Scaladocs */
 
+lazy val root = project
+  .in(file("."))
+  .aggregate(vectorpipe, examples)
+  .settings(commonSettings, vpExtraSettings)
+
 /* Main project */
 lazy val vectorpipe = project
-  .in(file("."))
+  .in(file("core"))
   .settings(commonSettings, vpExtraSettings, release)
 
+/* Example projects */
+lazy val examples = project
+  .in(file("examples"))
+  .settings(commonSettings, vpExtraSettings)
+  .dependsOn(vectorpipe)
+
 /* Benchmarking suite.
  * Benchmarks can be executed by first switching to the `bench` project and then by running:
       jmh:run -t 1 -f 1 -wi 5 -i 5 .*Bench.*
@@ -162,7 +173,7 @@ lazy val bench = project
   .dependsOn(vectorpipe)
   .enablePlugins(JmhPlugin)
 
-
+onLoad in Global ~= (_ andThen ("project vectorpipe" :: _))
 
 
 // assemblyShadeRules in assembly := {

…che.spark.sql.sources.DataSourceRegister …che.spark.sql.sources.DataSourceRegistersrc/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister renamed to core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister renamed to core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister

@@ -15,6 +15,8 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.StringType
 import org.locationtech.jts.{geom => jts}
 
+import scala.reflect.ClassTag
+
 object VectorPipe {
 
   /** Vectortile conversion options.
@@ -46,7 +48,10 @@ object VectorPipe {
     def forAllZoomsWithSrcProjection(zoom: Int, crs: CRS) = Options(zoom, Some(0), crs, None)
   }
 
-  def apply(input: DataFrame, pipeline: vectortile.Pipeline, options: Options): Unit = {
+  def apply[T: ClassTag](input: DataFrame, pipeline: vectortile.Pipeline, options: Options): Unit = {
+    import input.sparkSession.implicits._
+    import vectorpipe.encoders._
+
     val geomColumn = pipeline.geometryColumn
     assert(input.columns.contains(geomColumn) &&
            input.schema(geomColumn).dataType.isInstanceOf[org.apache.spark.sql.jts.AbstractGeometryUDT[jts.Geometry]],
@@ -74,46 +79,49 @@ object VectorPipe {
         SpatialKey(k.col / 2, k.row / 2) }.toSeq
     }
 
-    def generateVectorTiles[G <: Geometry](df: DataFrame, level: LayoutLevel): RDD[(SpatialKey, VectorTile)] = {
+    def generateVectorTiles[G <: Geometry](df: DataFrame, level: LayoutLevel): Dataset[(SpatialKey, Array[Byte])] = {
       val zoom = level.zoom
-      val clip = udf { (g: jts.Geometry, key: GenericRowWithSchema) =>
-        val k = getSpatialKey(key)
-        pipeline.clip(g, k, level)
-      }
 
-      val selectedGeometry = pipeline
-        .select(df, zoom, keyColumn)
+      val selectedGeometry = pipeline.select match {
+        case None => df
+        case Some(select) => select(df, zoom, keyColumn)
+      }
 
-      val clipped = selectedGeometry
+      val keyed = selectedGeometry
         .withColumn(keyColumn, explode(col(keyColumn)))
-        .repartition(col(keyColumn)) // spread copies of possibly ill-tempered geometries around cluster prior to clipping
-        .withColumn(geomColumn, clip(col(geomColumn), col(keyColumn)))
+
+      val clipped = pipeline.clip match {
+        case None => keyed
+        case Some(clipper) =>
+          val clip = udf { (g: jts.Geometry, key: GenericRowWithSchema) =>
+            val k = getSpatialKey(key)
+            clipper(g, k, level)
+          }
+          val toClip = keyed.repartition(col(keyColumn)) // spread copies of possibly ill-tempered geometries around cluster prior to clipping
+          toClip.withColumn(geomColumn, clip(col(geomColumn), col(keyColumn)))
+      }
 
       pipeline.layerMultiplicity match {
         case SingleLayer(layerName) =>
           clipped
-            .rdd
-            .map { r => (getSpatialKey(r, keyColumn), pipeline.pack(r, zoom)) }
-            .groupByKey
-            .map { case (key, feats) =>
+            .map { r => SingleLayerEntry(getSpatialKey(r, keyColumn), pipeline.pack(r, zoom)) }
+            .groupByKey(_.key)
+            .mapGroups { (key: SpatialKey, sleIter: Iterator[SingleLayerEntry]) =>
                val ex = level.layout.mapTransform.keyToExtent(key)
-               key -> buildVectorTile(feats, layerName, ex, options.tileResolution, options.orderAreas)
+               key -> buildVectorTile(sleIter.map(_.feature).toIterable, layerName, ex, options.tileResolution, options.orderAreas).toBytes
             }
         case LayerNamesInColumn(layerNameCol) =>
           assert(selectedGeometry.schema(layerNameCol).dataType == StringType,
                  s"layerMultiplicity=${pipeline.layerMultiplicity} requires String-type column of name ${layerNameCol}")
+
           clipped
-            .rdd
-            .map { r => (getSpatialKey(r, keyColumn), r.getAs[String](layerNameCol) -> pipeline.pack(r, zoom)) }
-            .groupByKey
-            .mapPartitions{ iter: Iterator[(SpatialKey, Iterable[(String, VectorTileFeature[Geometry])])] =>
-              iter.map{ case (key, groupedFeatures) => {
-                val layerFeatures: Map[String, Iterable[VectorTileFeature[Geometry]]] =
-                  groupedFeatures.groupBy(_._1).mapValues(_.map(_._2))
-                val ex = level.layout.mapTransform.keyToExtent(key)
-                key -> buildVectorTile(layerFeatures, ex, options.tileResolution, options.orderAreas)
-              }}
-          }
+            .map { r => MultipleLayerEntry(getSpatialKey(r, keyColumn), r.getAs[String](layerNameCol), pipeline.pack(r, zoom)) }
+            .groupByKey(_.key)
+            .mapGroups{ (key: SpatialKey, iter: Iterator[MultipleLayerEntry]) =>
+              val ex = level.layout.mapTransform.keyToExtent(key)
+              val layerFeatures = iter.toSeq.groupBy(_.layer).mapValues(_.map(_.feature))
+              key -> buildVectorTile(layerFeatures, ex, options.tileResolution, options.orderAreas).toBytes
+            }
       }
     }
 
@@ -134,16 +142,30 @@ object VectorPipe {
         } else {
           df
         }
-      val simplify = udf { g: jts.Geometry => pipeline.simplify(g, level.layout) }
-      val reduced = pipeline
-        .reduce(working, level, keyColumn)
-      val prepared = reduced
-        .withColumn(geomColumn, simplify(col(geomColumn)))
-      val vts = generateVectorTiles(prepared, level)
+
+      val reduced = pipeline.reduce match {
+        case None => working
+        case Some(reduce) => reduce(working, level, keyColumn)
+      }
+
+      val simplified = pipeline.simplify match {
+        case None => reduced
+        case Some(simplifier) =>
+          val simplify = udf { g: jts.Geometry => simplifier(g, level.layout) }
+          reduced.withColumn(geomColumn, simplify(col(geomColumn)))
+      }
+
+      val vts = generateVectorTiles(simplified, level)
       saveVectorTiles(vts, zoom, pipeline.baseOutputURI)
-      prepared.withColumn(keyColumn, reduceKeys(col(keyColumn)))
+
+      simplified.withColumn(keyColumn, reduceKeys(col(keyColumn)))
     }
 
   }
 
+  private case class SingleLayerEntry(key: SpatialKey, feature: VectorTileFeature[Geometry])
+  private case class MultipleLayerEntry(key: SpatialKey, layer: String, feature: VectorTileFeature[Geometry])
+
+  private implicit def sleEncoder: Encoder[SingleLayerEntry] = Encoders.kryo[SingleLayerEntry]
+  private implicit def mleEncoder: Encoder[MultipleLayerEntry] = Encoders.kryo[MultipleLayerEntry]
 }

…c/main/resources/microsite/data/menu.yml …c/main/resources/microsite/data/menu.ymlsrc/main/resources/microsite/data/menu.yml renamed to core/src/main/resources/microsite/data/menu.yml src/main/resources/microsite/data/menu.yml renamed to core/src/main/resources/microsite/data/menu.yml

@@ -0,0 +1,22 @@
+package vectorpipe.encoders
+
+import geotrellis.vector._
+import geotrellis.vectortile._
+import org.apache.spark.sql.{Encoder, Encoders}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+
+object GTEncoders {
+  implicit def gtGeometryEncoder: Encoder[Geometry] = Encoders.kryo[Geometry]
+  implicit def gtPointEncoder: Encoder[Point] = ExpressionEncoder()
+  implicit def gtMultiPointEncoder: Encoder[MultiPoint] = ExpressionEncoder()
+  implicit def gtLineEncoder: Encoder[Line] = ExpressionEncoder()
+  implicit def gtMultiLineEncoder: Encoder[MultiLine] = ExpressionEncoder()
+  implicit def gtPolygonEncoder: Encoder[Polygon] = ExpressionEncoder()
+  implicit def gtMultiPolygonEncoder: Encoder[MultiPolygon] = ExpressionEncoder()
+
+  implicit def gtFeatureEncoder[G <: Geometry, D](implicit ev1: Encoder[G], ev2: Encoder[D]): Encoder[Feature[G, D]] = Encoders.kryo[Feature[G, D]]
+
+  implicit def gtVectorTileEncoder: Encoder[VectorTile] = Encoders.kryo[VectorTile]
+  //implicit def gtLayerEncoder: Encoder[Layer] = Encoders.javaSerialization[Layer]
+  //implicit def gtStrictLayerEncoder: Encoder[StrictLayer] = Encoders.kryo[StrictLayer]
+}

src/main/scala/vectorpipe/OSM.scala core/src/main/scala/vectorpipe/OSM.scalasrc/main/scala/vectorpipe/OSM.scala renamed to core/src/main/scala/vectorpipe/OSM.scala

@@ -182,7 +182,7 @@ package object internal {
 
       // when an element has been deleted, it doesn't include any tags; use a window function to retrieve the last tags
       // present and use those
-      history
+      frame
         .where('type === "relation")
         .repartition('id)
         .select(