Upgraded to Spark 2.4.4

Author: metasim

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

@@ -22,12 +22,11 @@ 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.locationtech.rasterframes.expressions.generators.ExplodeTiles
+import org.apache.spark.sql._
+import org.apache.spark.sql.functions._
 import org.openjdk.jmh.annotations._
+
 /**
  *
  * @author sfitch
@@ -37,32 +36,33 @@ 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("uint16ud255"))
+  @Param(Array("uint8", "uint16ud255", "float32", "float64"))
   var cellTypeName: String = _
 
   @Param(Array("256"))
   var tileSize: Int = _
 
-  @Param(Array("2000"))
+  @Param(Array("100"))
   var numTiles: Int = _
 
   @transient
-  var tiles: Array[InternalRow] = _
-
-  var exploder: ExplodeTiles = _
+  var tiles: DataFrame = _
 
   @Setup(Level.Trial)
   def setupData(): Unit = {
-    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))
+    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()
   }
+
   @Benchmark
   def tileExplode() = {
-    for(t <- tiles)
-      exploder.eval(t)
+    tiles.select(rf_explode_tiles($"tile")).count()
   }
 }

build.sbt

@@ -32,7 +32,10 @@ lazy val root = project
   .withId("RasterFrames")
   .aggregate(core, datasource, pyrasterframes, experimental)
   .enablePlugins(RFReleasePlugin)
-  .settings(publish / skip := true)
+  .settings(
+    publish / skip := true,
+    clean := clean.dependsOn(`rf-notebook`/clean).value
+  )
 
 lazy val `rf-notebook` = project
   .dependsOn(pyrasterframes)

core/src/main/scala/org/locationtech/rasterframes/expressions/generators/ExplodeTiles.scala

@@ -24,8 +24,8 @@ package org.locationtech.rasterframes.expressions.generators
 import geotrellis.raster._
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.codegen.{BufferHolder, CodegenFallback, UnsafeRowWriter}
-import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, GenericInternalRow}
 import org.apache.spark.sql.types._
 import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.DynamicExtractors
@@ -87,17 +87,14 @@ case class ExplodeTiles(
       cfor(0)(_ < rows, _ + 1) { row =>
         cfor(0)(_ < cols, _ + 1) { col =>
           val rowIndex = row * cols + col
-          val outRow = new UnsafeRow(numOutCols)
-          val buffer = new BufferHolder(outRow)
-          val writer = new UnsafeRowWriter(buffer, numOutCols)
-          writer.write(0, col)
-          writer.write(1, row)
+          val outCols = Array.ofDim[Any](numOutCols)
+          outCols(0) = col
+          outCols(1) = row
           cfor(0)(_ < tiles.length, _ + 1) { index =>
             val tile = tiles(index)
-            val cell: Double = if (tile == null) doubleNODATA else tile.getDouble(col, row)
-            writer.write(index + 2, cell)
+            outCols(index + 2) = if(tile == null) doubleNODATA else tile.getDouble(col, row)
           }
-          retval(rowIndex) = outRow
+          retval(rowIndex) = new GenericInternalRow(outCols)
         }
       }
       if(sampleFraction > 0.0 && sampleFraction < 1.0) sample(retval)

core/src/main/scala/org/locationtech/rasterframes/expressions/package.scala

@@ -36,7 +36,7 @@ import org.locationtech.rasterframes.expressions.tilestats._
 import org.locationtech.rasterframes.expressions.transformers._
 
 import scala.reflect.runtime.universe._
-import scala.util.Try
+
 /**
  * Module of Catalyst expressions for efficiently working with tiles.
  *
@@ -53,8 +53,7 @@ package object expressions {
   private[expressions]
   def udfexpr[RT: TypeTag, A1: TypeTag](name: String, f: A1 => RT): Expression => ScalaUDF = (child: Expression) => {
     val ScalaReflection.Schema(dataType, nullable) = ScalaReflection.schemaFor[RT]
-    val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType :: Nil).toOption
-    ScalaUDF(f, dataType, Seq(child),  inputTypes.getOrElse(Nil), nullable = nullable, udfName = Some(name))
+    ScalaUDF(f, dataType, Seq(child), Seq(true), nullable = nullable, udfName = Some(name))
   }
 
   def register(sqlContext: SQLContext): Unit = {

core/src/main/scala/org/locationtech/rasterframes/rasterframes.scala

@@ -31,15 +31,13 @@ import org.locationtech.geomesa.spark.jts.DataFrameFunctions
 import org.locationtech.rasterframes.encoders.StandardEncoders
 import org.locationtech.rasterframes.extensions.Implicits
 import org.locationtech.rasterframes.model.TileDimensions
-import org.locationtech.rasterframes.util.ZeroSevenCompatibilityKit
 import org.slf4j.LoggerFactory
 import shapeless.tag.@@
 
 import scala.reflect.runtime.universe._
 
 package object rasterframes extends StandardColumns
   with RasterFunctions
-  with ZeroSevenCompatibilityKit.RasterFunctions
   with Implicits
   with rasterframes.jts.Implicits
   with StandardEncoders
@@ -81,7 +79,6 @@ package object rasterframes extends StandardColumns
     }
 
     rf.register(sqlContext)
-    ZeroSevenCompatibilityKit.register(sqlContext)
     rasterframes.functions.register(sqlContext)
     rasterframes.expressions.register(sqlContext)
     rasterframes.rules.register(sqlContext)