Fix for SparkML memory pressure issues

bench/archive/jmh-results-20190912131522.json

@@ -0,0 +1,108 @@
+[
+    {
+        "jmhVersion" : "1.21",
+        "benchmark" : "org.locationtech.rasterframes.bench.CellTypeBench.fromRow",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 1,
+        "jvm" : "/Library/Java/JavaVirtualMachines/jdk1.8.0_171.jdk/Contents/Home/jre/bin/java",
+        "jvmArgs" : [
+            "-Xmx4g"
+        ],
+        "jdkVersion" : "1.8.0_171",
+        "vmName" : "Java HotSpot(TM) 64-Bit Server VM",
+        "vmVersion" : "25.171-b11",
+        "warmupIterations" : 8,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "10 s",
+        "measurementBatchSize" : 1,
+        "primaryMetric" : {
+            "score" : 2.835836726694286,
+            "scoreError" : 0.19270422040266105,
+            "scoreConfidence" : [
+                2.643132506291625,
+                3.028540947096947
+            ],
+            "scorePercentiles" : {
+                "0.0" : 2.795428349144724,
+                "50.0" : 2.833201108807787,
+                "90.0" : 2.918007291796378,
+                "95.0" : 2.918007291796378,
+                "99.0" : 2.918007291796378,
+                "99.9" : 2.918007291796378,
+                "99.99" : 2.918007291796378,
+                "99.999" : 2.918007291796378,
+                "99.9999" : 2.918007291796378,
+                "100.0" : 2.918007291796378
+            },
+            "scoreUnit" : "us/op",
+            "rawData" : [
+                [
+                    2.795428349144724,
+                    2.79552918770274,
+                    2.837017696019802,
+                    2.918007291796378,
+                    2.833201108807787
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.21",
+        "benchmark" : "org.locationtech.rasterframes.bench.CellTypeBench.intoRow",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 1,
+        "jvm" : "/Library/Java/JavaVirtualMachines/jdk1.8.0_171.jdk/Contents/Home/jre/bin/java",
+        "jvmArgs" : [
+            "-Xmx4g"
+        ],
+        "jdkVersion" : "1.8.0_171",
+        "vmName" : "Java HotSpot(TM) 64-Bit Server VM",
+        "vmVersion" : "25.171-b11",
+        "warmupIterations" : 8,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "10 s",
+        "measurementBatchSize" : 1,
+        "primaryMetric" : {
+            "score" : 0.35239272354175755,
+            "scoreError" : 0.007379508933042929,
+            "scoreConfidence" : [
+                0.3450132146087146,
+                0.3597722324748005
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.34961161284617265,
+                "50.0" : 0.3525031701325926,
+                "90.0" : 0.3550145438728285,
+                "95.0" : 0.3550145438728285,
+                "99.0" : 0.3550145438728285,
+                "99.9" : 0.3550145438728285,
+                "99.99" : 0.3550145438728285,
+                "99.999" : 0.3550145438728285,
+                "99.9999" : 0.3550145438728285,
+                "100.0" : 0.3550145438728285
+            },
+            "scoreUnit" : "us/op",
+            "rawData" : [
+                [
+                    0.3550145438728285,
+                    0.3522314263912252,
+                    0.3525031701325926,
+                    0.34961161284617265,
+                    0.3526028644659687
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    }
+]
+
+

bench/archive/jmh-results-20190912133231.json

@@ -0,0 +1,108 @@
+[
+    {
+        "jmhVersion" : "1.21",
+        "benchmark" : "org.locationtech.rasterframes.bench.CellTypeBench.fromRow",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 1,
+        "jvm" : "/Library/Java/JavaVirtualMachines/jdk1.8.0_171.jdk/Contents/Home/jre/bin/java",
+        "jvmArgs" : [
+            "-Xmx4g"
+        ],
+        "jdkVersion" : "1.8.0_171",
+        "vmName" : "Java HotSpot(TM) 64-Bit Server VM",
+        "vmVersion" : "25.171-b11",
+        "warmupIterations" : 8,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "10 s",
+        "measurementBatchSize" : 1,
+        "primaryMetric" : {
+            "score" : 0.09014145135858256,
+            "scoreError" : 0.002394012003996672,
+            "scoreConfidence" : [
+                0.0877474393545859,
+                0.09253546336257923
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.08936698388038906,
+                "50.0" : 0.08989743516282808,
+                "90.0" : 0.09083590280173164,
+                "95.0" : 0.09083590280173164,
+                "99.0" : 0.09083590280173164,
+                "99.9" : 0.09083590280173164,
+                "99.99" : 0.09083590280173164,
+                "99.999" : 0.09083590280173164,
+                "99.9999" : 0.09083590280173164,
+                "100.0" : 0.09083590280173164
+            },
+            "scoreUnit" : "us/op",
+            "rawData" : [
+                [
+                    0.08989743516282808,
+                    0.09072300570225225,
+                    0.08988392924571173,
+                    0.09083590280173164,
+                    0.08936698388038906
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    },
+    {
+        "jmhVersion" : "1.21",
+        "benchmark" : "org.locationtech.rasterframes.bench.CellTypeBench.intoRow",
+        "mode" : "avgt",
+        "threads" : 1,
+        "forks" : 1,
+        "jvm" : "/Library/Java/JavaVirtualMachines/jdk1.8.0_171.jdk/Contents/Home/jre/bin/java",
+        "jvmArgs" : [
+            "-Xmx4g"
+        ],
+        "jdkVersion" : "1.8.0_171",
+        "vmName" : "Java HotSpot(TM) 64-Bit Server VM",
+        "vmVersion" : "25.171-b11",
+        "warmupIterations" : 8,
+        "warmupTime" : "10 s",
+        "warmupBatchSize" : 1,
+        "measurementIterations" : 5,
+        "measurementTime" : "10 s",
+        "measurementBatchSize" : 1,
+        "primaryMetric" : {
+            "score" : 0.0804223243178353,
+            "scoreError" : 0.003232629425106684,
+            "scoreConfidence" : [
+                0.07718969489272862,
+                0.08365495374294199
+            ],
+            "scorePercentiles" : {
+                "0.0" : 0.07979611499160112,
+                "50.0" : 0.08015109899040546,
+                "90.0" : 0.08189719727590557,
+                "95.0" : 0.08189719727590557,
+                "99.0" : 0.08189719727590557,
+                "99.9" : 0.08189719727590557,
+                "99.99" : 0.08189719727590557,
+                "99.999" : 0.08189719727590557,
+                "99.9999" : 0.08189719727590557,
+                "100.0" : 0.08189719727590557
+            },
+            "scoreUnit" : "us/op",
+            "rawData" : [
+                [
+                    0.08189719727590557,
+                    0.08020909063591224,
+                    0.07979611499160112,
+                    0.08005811969535213,
+                    0.08015109899040546
+                ]
+            ]
+        },
+        "secondaryMetrics" : {
+        }
+    }
+]
+
+

bench/src/main/scala/org/locationtech/rasterframes/bench/CellTypeBench.scala

@@ -0,0 +1,52 @@
+/*
+ * This software is licensed under the Apache 2 license, quoted below.
+ *
+ * Copyright 2019 Astraea, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *     [http://www.apache.org/licenses/LICENSE-2.0]
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.locationtech.rasterframes.bench
+import java.util.concurrent.TimeUnit
+
+import geotrellis.raster.{CellType, DoubleUserDefinedNoDataCellType, IntUserDefinedNoDataCellType}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.locationtech.rasterframes.encoders.CatalystSerializer._
+import org.openjdk.jmh.annotations._
+
+@BenchmarkMode(Array(Mode.AverageTime))
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+class CellTypeBench {
+  var row: InternalRow = _
+  var ct: CellType = _
+  @Setup(Level.Trial)
+  def setupData(): Unit = {
+    ct = IntUserDefinedNoDataCellType(scala.util.Random.nextInt())
+    val o: CellType = DoubleUserDefinedNoDataCellType(scala.util.Random.nextDouble())
+    row = o.toInternalRow
+  }
+
+  @Benchmark
+  def fromRow(): CellType = {
+    row.to[CellType]
+  }
+
+  @Benchmark
+  def intoRow(): InternalRow = {
+    ct.toInternalRow
+  }
+}

bench/src/main/scala/org/locationtech/rasterframes/bench/TileExplodeBench.scala

@@ -22,11 +22,12 @@ package org.locationtech.rasterframes.bench
 
 import java.util.concurrent.TimeUnit
 
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.BoundReference
+import org.apache.spark.sql.rf.TileUDT
 import org.locationtech.rasterframes._
-import org.apache.spark.sql._
-import org.apache.spark.sql.functions._
+import org.locationtech.rasterframes.expressions.generators.ExplodeTiles
 import org.openjdk.jmh.annotations._
-
 /**
  *
  * @author sfitch
@@ -36,33 +37,32 @@ import org.openjdk.jmh.annotations._
 @State(Scope.Benchmark)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
 class TileExplodeBench extends SparkEnv {
-  import spark.implicits._
 
-  @Param(Array("uint8", "uint16ud255", "float32", "float64"))
+  //@Param(Array("uint8", "uint16ud255", "float32", "float64"))
+  @Param(Array("uint16ud255"))
   var cellTypeName: String = _
 
   @Param(Array("256"))
   var tileSize: Int = _
 
-  @Param(Array("100"))
+  @Param(Array("2000"))
   var numTiles: Int = _
 
   @transient
-  var tiles: DataFrame = _
+  var tiles: Array[InternalRow] = _
+
+  var exploder: ExplodeTiles = _
 
   @Setup(Level.Trial)
   def setupData(): Unit = {
-    tiles = Seq.fill(numTiles)(randomTile(tileSize, tileSize, cellTypeName))
-      .toDF("tile").repartition(10)
-  }
-
-  @Benchmark
-  def arrayExplode() = {
-    tiles.select(posexplode(rf_tile_to_array_double($"tile"))).count()
+    tiles = Array.fill(numTiles)(randomTile(tileSize, tileSize, cellTypeName))
+        .map(t => InternalRow(TileUDT.tileSerializer.toInternalRow(t)))
+    val expr = BoundReference(0, TileType, true)
+    exploder = new ExplodeTiles(1.0, None, Seq(expr))
   }
-
   @Benchmark
   def tileExplode() = {
-    tiles.select(rf_explode_tiles($"tile")).count()
+    for(t <- tiles)
+      exploder.eval(t)
   }
 }

core/src/main/scala/org/apache/spark/sql/rf/RasterSourceUDT.scala

@@ -73,7 +73,7 @@ object RasterSourceUDT {
 
   implicit val rasterSourceSerializer: CatalystSerializer[RasterSource] = new CatalystSerializer[RasterSource] {
 
-    override def schema: StructType = StructType(Seq(
+    override val schema: StructType = StructType(Seq(
       StructField("raster_source_kryo", BinaryType, false)
     ))
 

core/src/main/scala/org/apache/spark/sql/rf/TileUDT.scala

@@ -74,7 +74,7 @@ case object TileUDT  {
 
   implicit def tileSerializer: CatalystSerializer[Tile] = new CatalystSerializer[Tile] {
 
-    override def schema: StructType = StructType(Seq(
+    override val schema: StructType = StructType(Seq(
       StructField("cell_context", schemaOf[TileDataContext], false),
       StructField("cell_data", schemaOf[Cells], false)
     ))