Added ability to write unstructured RasterFrame to GeoTiff.

Author: metasim

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

@@ -154,4 +154,9 @@ object CatalystSerializer extends StandardSerializers {
   implicit class WithFromRow(val r: Row) extends AnyVal {
     def to[T >: Null: CatalystSerializer]: T = if (r == null) null else CatalystSerializer[T].fromRow(r)
   }
+
+  implicit class WithTypeConformity(val left: DataType) extends AnyVal {
+    def conformsTo[T >: Null: CatalystSerializer]: Boolean =
+      org.apache.spark.sql.rf.WithTypeConformity(left).conformsTo(schemaOf[T])
+  }
 }

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

@@ -24,7 +24,7 @@ package org.locationtech.rasterframes.expressions
 import geotrellis.raster.{CellGrid, Tile}
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.rf.{TileUDT, _}
+import org.apache.spark.sql.rf.{TileUDT, RasterSourceUDT}
 import org.apache.spark.sql.types._
 import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.model.TileContext
@@ -38,7 +38,7 @@ object DynamicExtractors {
     case _: TileUDT =>
       (row: InternalRow) =>
         (row.to[Tile](TileUDT.tileSerializer), None)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => {
         val prt = row.to[ProjectedRasterTile]
         (prt, Some(TileContext(prt)))
@@ -48,7 +48,7 @@ object DynamicExtractors {
   lazy val rowTileExtractor: PartialFunction[DataType, Row => (Tile, Option[TileContext])] = {
     case _: TileUDT =>
       (row: Row) =>  (row.to[Tile](TileUDT.tileSerializer), None)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: Row) => {
         val prt = row.to[ProjectedRasterTile]
         (prt, Some(TileContext(prt)))
@@ -59,9 +59,9 @@ object DynamicExtractors {
   lazy val projectedRasterLikeExtractor: PartialFunction[DataType, InternalRow ⇒ ProjectedRasterLike] = {
     case _: RasterSourceUDT ⇒
       (row: InternalRow) => row.to[RasterSource](RasterSourceUDT.rasterSourceSerializer)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => row.to[ProjectedRasterTile]
-    case t if t.conformsTo(schemaOf[RasterRef]) =>
+    case t if t.conformsTo[RasterRef] =>
       (row: InternalRow) => row.to[RasterRef]
   }
 
@@ -71,9 +71,9 @@ object DynamicExtractors {
       (row: InternalRow) => row.to[Tile](TileUDT.tileSerializer)
     case _: RasterSourceUDT =>
       (row: InternalRow) => row.to[RasterSource](RasterSourceUDT.rasterSourceSerializer)
-    case t if t.conformsTo(schemaOf[RasterRef]) ⇒
+    case t if t.conformsTo[RasterRef] ⇒
       (row: InternalRow) => row.to[RasterRef]
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => row.to[ProjectedRasterTile]
   }
 

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

@@ -30,7 +30,6 @@ import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{ScalaUDF, _}
 import org.apache.spark.sql.jts.AbstractGeometryUDT
-import org.apache.spark.sql.rf.WithTypeConformity
 import org.apache.spark.sql.types._
 import org.locationtech.geomesa.spark.jts.udf.SpatialRelationFunctions._
 
@@ -48,7 +47,7 @@ abstract class SpatialRelation extends BinaryExpression
       case r: InternalRow ⇒
         expr.dataType match {
           case udt: AbstractGeometryUDT[_] ⇒ udt.deserialize(r)
-          case dt if dt.conformsTo(schemaOf[Extent]) =>
+          case dt if dt.conformsTo[Extent] =>
             val extent = r.to[Extent]
             extent.jtsGeom
         }

core/src/main/scala/org/locationtech/rasterframes/expressions/aggregates/TileRasterizerAggregate.scala

@@ -21,16 +21,17 @@
 
 package org.locationtech.rasterframes.expressions.aggregates
 
-import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import geotrellis.proj4.CRS
+import geotrellis.raster.reproject.Reproject
 import geotrellis.raster.resample.ResampleMethod
 import geotrellis.raster.{ArrayTile, CellType, Raster, Tile}
+import geotrellis.spark.TileLayerMetadata
 import geotrellis.vector.Extent
 import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
 import org.apache.spark.sql.types.{DataType, StructField, StructType}
 import org.apache.spark.sql.{Column, Row, TypedColumn}
-import geotrellis.raster.reproject.Reproject
+import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggregate.ProjectedRasterDefinition
 
 /**
@@ -87,10 +88,21 @@ class TileRasterizerAggregate(prd: ProjectedRasterDefinition) extends UserDefine
 }
 
 object TileRasterizerAggregate {
-  val nodeName = "tile_rasterizer_aggregate"
+  val nodeName = "rf_tile_rasterizer_aggregate"
   /**  Convenience grouping of  parameters needed for running aggregate. */
   case class ProjectedRasterDefinition(cols: Int, rows: Int, cellType: CellType, crs: CRS, extent: Extent, sampler: ResampleMethod = ResampleMethod.DEFAULT)
 
+  object ProjectedRasterDefinition {
+    def apply(tlm: TileLayerMetadata[_]): ProjectedRasterDefinition = apply(tlm, ResampleMethod.DEFAULT)
+
+    def apply(tlm: TileLayerMetadata[_], sampler: ResampleMethod): ProjectedRasterDefinition = {
+      val actualSize = tlm.layout.toRasterExtent().gridBoundsFor(tlm.extent)
+      val cols = actualSize.width
+      val rows = actualSize.height
+      new ProjectedRasterDefinition(cols, rows, tlm.cellType, tlm.crs, tlm.extent, sampler)
+    }
+}
+
   def apply(prd: ProjectedRasterDefinition, crsCol: Column, extentCol: Column, tileCol: Column): TypedColumn[Any, Raster[Tile]] =
     new TileRasterizerAggregate(prd)(crsCol, extentCol, tileCol).as(nodeName).as[Raster[Tile]]
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/transformers/SetCellType.scala

@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure,
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{BinaryExpression, Expression}
 import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.rf._
+import org.apache.spark.sql.rf.{TileUDT, WithTypeConformity}
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, TypedColumn}
 import org.apache.spark.unsafe.types.UTF8String

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

@@ -21,20 +21,23 @@
 
 package org.locationtech.rasterframes.extensions
 
-import org.locationtech.rasterframes.StandardColumns._
-import org.locationtech.rasterframes.util._
-import org.locationtech.rasterframes.RasterFrameLayer
-import geotrellis.raster.Tile
+import geotrellis.proj4.CRS
+import geotrellis.raster.{MultibandTile, Tile}
 import geotrellis.spark.io._
 import geotrellis.spark.{SpaceTimeKey, SpatialComponent, SpatialKey, TemporalKey, TileLayerMetadata}
 import geotrellis.util.MethodExtensions
+import geotrellis.vector.{Extent, ProjectedExtent}
+import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.types.{MetadataBuilder, StructField}
 import org.apache.spark.sql.{Column, DataFrame, TypedColumn}
-import spray.json.JsonFormat
+import org.locationtech.rasterframes.{MetadataKeys, RasterFrameLayer}
+import org.locationtech.rasterframes.StandardColumns._
+import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.encoders.StandardEncoders._
 import org.locationtech.rasterframes.expressions.DynamicExtractors
-import org.locationtech.rasterframes.MetadataKeys
+import org.locationtech.rasterframes.util._
+import spray.json.JsonFormat
 
 import scala.util.Try
 
@@ -91,7 +94,19 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
   def tileColumns: Seq[Column] =
     self.schema.fields
       .filter(f => DynamicExtractors.tileExtractor.isDefinedAt(f.dataType))
-      .map(f ⇒ self.col(f.name).as[Tile])
+      .map(f ⇒ self.col(f.name))
+
+  /** Get the columns that look like `Extent`s. */
+  def extentColumns: Seq[Column] =
+    self.schema.fields
+      .filter(_.dataType.conformsTo[Extent])
+      .map(f => self.col(f.name))
+
+  /** Get the columns that look like `CRS`s. */
+  def crsColumns: Seq[Column] =
+    self.schema.fields
+      .filter(_.dataType.conformsTo[CRS])
+      .map(f => self.col(f.name))
 
   /** Get the columns that are not of type `Tile` */
   def notTileColumns: Seq[Column] =
@@ -135,38 +150,6 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
   def withPrefixedColumnNames(prefix: String): DF =
     self.columns.foldLeft(self)((df, c) ⇒ df.withColumnRenamed(c, s"$prefix$c").asInstanceOf[DF])
 
-  /** Converts this DataFrame to a RasterFrameLayer after ensuring it has:
-   *
-   * <ol type="a">
-   * <li>a space or space-time key column
-   * <li>one or more tile columns
-   * <li>tile layout metadata
-   * <ol>
-   *
-   * If any of the above are violated, and [[IllegalArgumentException]] is thrown.
-   *
-   * @return validated RasterFrameLayer
-   * @throws IllegalArgumentException when constraints are not met.
-   */
-  @throws[IllegalArgumentException]
-  def asLayer: RasterFrameLayer = {
-    val potentialRF = certifyRasterframe(self)
-
-    require(
-      potentialRF.findSpatialKeyField.nonEmpty,
-      "A RasterFrameLayer requires a column identified as a spatial key"
-    )
-
-    require(potentialRF.tileColumns.nonEmpty, "A RasterFrameLayer requires at least one tile column")
-
-    require(
-      Try(potentialRF.tileLayerMetadata).isSuccess,
-      "A RasterFrameLayer requires embedded TileLayerMetadata"
-    )
-
-    potentialRF
-  }
-
   /**
     * Performs a jeft join on the dataframe `right` to this one, reprojecting and merging tiles as necessary.
     * The operation is logically a "left outer" join, with the left side also determining the target CRS and extents.
@@ -218,6 +201,42 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
   def rasterJoin(right: DataFrame, joinExpr: Column, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column): DataFrame =
     RasterJoin(self, right, joinExpr, leftExtent, leftCRS, rightExtent, rightCRS)
 
+
+  /** Layout contents of RasterFrame to a layer. Assumes CRS and extent columns exist. */
+  def toLayer(tlm: TileLayerMetadata[SpatialKey]): RasterFrameLayer = ReprojectToLayer(self, tlm)
+
+  /** Coerces this DataFrame to a RasterFrameLayer after ensuring it has:
+    *
+    * <ol type="a">
+    * <li>a space or space-time key column
+    * <li>one or more tile columns
+    * <li>tile layout metadata
+    * <ol>
+    *
+    * If any of the above are violated, and [[IllegalArgumentException]] is thrown.
+    *
+    * @return validated RasterFrameLayer
+    * @throws IllegalArgumentException when constraints are not met.
+    */
+  @throws[IllegalArgumentException]
+  def asLayer: RasterFrameLayer = {
+    val potentialRF = certifyRasterframe(self)
+
+    require(
+      potentialRF.findSpatialKeyField.nonEmpty,
+      "A RasterFrameLayer requires a column identified as a spatial key"
+    )
+
+    require(potentialRF.tileColumns.nonEmpty, "A RasterFrameLayer requires at least one tile column")
+
+    require(
+      Try(potentialRF.tileLayerMetadata).isSuccess,
+      "A RasterFrameLayer requires embedded TileLayerMetadata"
+    )
+
+    potentialRF
+  }
+
   /**
    * Convert DataFrame already in a uniform gridding into a RasterFrameLayer
    *
@@ -273,7 +292,7 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
    *
    * @return true if all constraints are fulfilled, false otherwise.
    */
-  def isLayer: Boolean = Try(asLayer).isSuccess
+  def isAlreadyLayer: Boolean = Try(asLayer).isSuccess
 
   /** Internal method for slapping the RasterFreameLayer seal of approval on a DataFrame.
    * Only call if if you are sure it has a spatial key and tile columns and TileLayerMetadata. */

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

@@ -20,48 +20,15 @@
  */
 
 package org.locationtech.rasterframes.extensions
-import geotrellis.proj4.CRS
-import geotrellis.raster._
-import geotrellis.raster.reproject.Reproject
-import geotrellis.vector.Extent
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.encoders.CatalystSerializer._
-import org.locationtech.rasterframes.model.TileDimensions
+import org.locationtech.rasterframes.functions.reproject_and_merge
 import org.locationtech.rasterframes.util._
 
 import scala.util.Random
 
 object RasterJoin {
-  private val projOpts = Reproject.Options.DEFAULT
-
-  val reproject_and_merge_f: (Row, Row, Seq[Tile], Seq[Row], Seq[Row], Row) => Tile = (leftExtentEnc: Row, leftCRSEnc: Row, tiles: Seq[Tile], rightExtentEnc: Seq[Row], rightCRSEnc: Seq[Row], leftDimsEnc: Row) => {
-    if (tiles.isEmpty) null
-    else {
-      require(tiles.length == rightExtentEnc.length && tiles.length == rightCRSEnc.length, "size mismatch")
-
-      val leftExtent = leftExtentEnc.to[Extent]
-      val leftDims = leftDimsEnc.to[TileDimensions]
-      val leftCRS = leftCRSEnc.to[CRS]
-      val rightExtents = rightExtentEnc.map(_.to[Extent])
-      val rightCRSs = rightCRSEnc.map(_.to[CRS])
-
-      val cellType = tiles.map(_.cellType).reduceOption(_ union _).getOrElse(tiles.head.cellType)
-
-      val dest: Tile = ArrayTile.empty(cellType, leftDims.cols, leftDims.rows)
-      //is there a GT function to do all this?
-      tiles.zip(rightExtents).zip(rightCRSs).map {
-        case ((tile, extent), crs) =>
-          tile.reproject(extent, crs, leftCRS, projOpts)
-      }.foldLeft(dest)((d, t) =>
-        d.merge(leftExtent, t.extent, t.tile, projOpts.method)
-      )
-    }
-  }
-
-  // NB: Don't be tempted to make this a `val`. Spark will barf if `withRasterFrames` hasn't been called first.
-  def reproject_and_merge = udf(reproject_and_merge_f).withName("reproject_and_merge")
 
   def apply(left: DataFrame, right: DataFrame): DataFrame = {
     val df = apply(left, right, left("extent"), left("crs"), right("extent"), right("crs"))

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

@@ -0,0 +1,49 @@
+/*
+ * 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.extensions
+
+import geotrellis.spark.{SpatialKey, TileLayerMetadata}
+import org.apache.spark.sql._
+import org.apache.spark.sql.functions.broadcast
+import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.util._
+object ReprojectToLayer {
+
+  def apply(df: DataFrame, tlm: TileLayerMetadata[SpatialKey]): RasterFrameLayer = {
+    // create a destination dataframe with crs and extend columns
+    // use RasterJoin to do the rest.
+    val gb = tlm.gridBounds
+    val crs = tlm.crs
+
+    val gridItems = for {
+      (col, row) <- gb.coordsIter
+      sk = SpatialKey(col, row)
+      e = tlm.mapTransform(sk)
+    } yield (sk, e, crs)
+
+    val dest = df.sparkSession.createDataFrame(gridItems.toSeq)
+      .toDF(SPATIAL_KEY_COLUMN.columnName, EXTENT_COLUMN.columnName, CRS_COLUMN.columnName)
+    val joined = RasterJoin(broadcast(dest), df)
+
+    joined.asLayer(SPATIAL_KEY_COLUMN, tlm)
+  }
+}