GeoTIFF writing post-walkthrough comments.

Author: metasim

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

@@ -94,9 +94,8 @@ object ProjectedLayerMetadataAggregate {
   // Ordering must match InputRecord schema
     new ProjectedLayerMetadataAggregate(destCRS, destDims)(extent, crs, cellType, tileSize).as[TileLayerMetadata[SpatialKey]]
 
-
   private[expressions]
-  case class InputRecord(extent: Extent, crs: CRS, cellType: CellType, tileSize: TileDimensions) { self ⇒
+  case class InputRecord(extent: Extent, crs: CRS, cellType: CellType, tileSize: TileDimensions) {
     def toBufferRecord(destCRS: CRS): BufferRecord = {
       val transform = Transform(crs, destCRS)
 

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

@@ -58,7 +58,7 @@ class TileRasterizerAggregate(prd: ProjectedRasterDefinition) extends UserDefine
   override def dataType: DataType = schemaOf[Raster[Tile]]
 
   override def initialize(buffer: MutableAggregationBuffer): Unit = {
-    buffer(0) = ArrayTile.empty(prd.cellType, prd.cols, prd.rows)
+    buffer(0) = ArrayTile.empty(prd.cellType, prd.totalCols, prd.totalRows)
   }
 
   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
@@ -90,13 +90,14 @@ class TileRasterizerAggregate(prd: ProjectedRasterDefinition) extends UserDefine
 object TileRasterizerAggregate {
   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)
+  case class ProjectedRasterDefinition(totalCols: Int, totalRows: 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)
+      // Try to determine the actual dimensions of our data coverage
+      val actualSize = tlm.layout.toRasterExtent().gridBoundsFor(tlm.extent) // <--- Do we have the math right here?
       val cols = actualSize.width
       val rows = actualSize.height
       new ProjectedRasterDefinition(cols, rows, tlm.cellType, tlm.crs, tlm.extent, sampler)

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

@@ -22,22 +22,20 @@
 package org.locationtech.rasterframes.extensions
 
 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 geotrellis.vector.Extent
 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 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.tiles.ProjectedRasterTile
 import org.locationtech.rasterframes.util._
+import org.locationtech.rasterframes.{MetadataKeys, RasterFrameLayer}
 import spray.json.JsonFormat
 
 import scala.util.Try

datasource/src/main/scala/org/locationtech/rasterframes/datasource/geotiff/GeoTiffDataSource.scala

@@ -24,19 +24,18 @@ package org.locationtech.rasterframes.datasource.geotiff
 import java.net.URI
 
 import _root_.geotrellis.proj4.CRS
+import _root_.geotrellis.raster._
 import _root_.geotrellis.raster.io.geotiff.compression._
 import _root_.geotrellis.raster.io.geotiff.tags.codes.ColorSpace
 import _root_.geotrellis.raster.io.geotiff.{GeoTiffOptions, MultibandGeoTiff, Tags, Tiled}
-import _root_.geotrellis.raster._
+import _root_.geotrellis.spark._
 import com.typesafe.scalalogging.LazyLogging
-import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, DataSourceRegister, RelationProvider}
 import org.apache.spark.sql._
-import org.apache.spark.sql.functions._
-import org.apache.spark.sql.rf.TileUDT
+import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, DataSourceRegister, RelationProvider}
 import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.datasource._
-import org.locationtech.rasterframes.expressions.aggregates.{ProjectedLayerMetadataAggregate, TileRasterizerAggregate}
 import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggregate.ProjectedRasterDefinition
+import org.locationtech.rasterframes.expressions.aggregates.{ProjectedLayerMetadataAggregate, TileRasterizerAggregate}
 import org.locationtech.rasterframes.model.{LazyCRS, TileDimensions}
 import org.locationtech.rasterframes.util._
 
@@ -96,32 +95,37 @@ class GeoTiffDataSource extends DataSourceRegister
       require(parameters.crs.nonEmpty, "A destination CRS must be provided")
       require(tileCols.nonEmpty, "need at least one tile column")
 
+      // Grab CRS to project into
       val destCRS = parameters.crs.get
 
-      val (extCol, crsCol) = {
+      // Select the anchoring Tile, Extent and CRS columns
+      val (extCol, crsCol, tileCol) = {
+        // Favor "ProjectedRaster" columns
         val prCols = df.projRasterColumns
         if(prCols.nonEmpty) {
-          (rf_extent(prCols.head), rf_crs(prCols.head))
+          (rf_extent(prCols.head), rf_crs(prCols.head), rf_tile(prCols.head))
         }
         else {
+          // If no "ProjectedRaster" column, look for single Extent and CRS columns.
           val crsCols = df.crsColumns
           require(crsCols.size == 1, "Exactly one CRS column must be in DataFrame")
           val extentCols = df.extentColumns
           require(extentCols.size == 1, "Exactly one Extent column must be in DataFrame")
-          (extentCols.head, crsCols.head)
+          (extentCols.head, crsCols.head, tileCols.head)
         }
       }
 
-      val tlm = df
+      // Scan table and constuct what the TileLayerMetadata would be in the specified destination CRS.
+      val tlm: TileLayerMetadata[SpatialKey] = df
         .select(ProjectedLayerMetadataAggregate(
-          destCRS, extCol, crsCol, rf_cell_type(tileCols.head), rf_dimensions(tileCols.head)
+          destCRS, extCol, crsCol, rf_cell_type(tileCol), rf_dimensions(tileCol)
         ))
         .first()
 
       val c = ProjectedRasterDefinition(tlm)
 
       val config = parameters.rasterDimensions.map { dims =>
-        c.copy(cols = dims.cols, rows = dims.rows)
+        c.copy(totalCols = dims.cols, totalRows = dims.rows)
       }.getOrElse(c)
 
       val aggs = tileCols

pyrasterframes/src/main/python/docs/raster-write.pymd

@@ -63,7 +63,7 @@ GeoTIFF is one of the most common file formats for spatial data, providing flexi
 
 One downside to GeoTIFF is that it is not a big data native format. To create a GeoTIFF all the data to be encoded has to be in the memory of one compute node (in Spark parlance, this is a "collect"), limiting it's maximum size substantially compared to that of a full cluster environment. When rendering GeoTIFFs in RasterFrames, you either need to specify the dimensions of the output raster, or be aware of how big the collected data will end up being.
 
-Fortunately, we can use the cluster computing capability to downlample the data into a more manageble saze. For sake of example, let's render a simple RGB overview image of our scene as a small raster:
+Fortunately, we can use the cluster computing capability to downlample the data (using nearest-neighbor) into a more manageble size. For sake of example, let's render a simple RGB overview image of our scene as a small raster, reprojecting it to latitude and longitude coordinates on the [WGS84](https://en.wikipedia.org/wiki/World_Geodetic_System) reference ellipsoid (aka [EPSG:4326](https://spatialreference.org/ref/epsg/4326/)):
 
 ```python write_geotiff
 import os.path
@@ -94,10 +94,9 @@ with rasterio.open(outfile) as src:
 
 ## GeoTrellis Layers
 
-[GeoTrellis][GeoTrellis] is one of the key libraries that RasterFrames builds upon. It provides a Scala language API to working with large raster data with Apache Spark. Ingesting raster data into a Layer is one of the key concepts for creating a dataset for processing on Spark. RasterFrames write data from an appropriate DataFrame into a GeoTrellis Layer.
-
-\[ More details see https://s22s.myjetbrains.com/youtrack/issue/RF-72 \]
+[GeoTrellis][GeoTrellis] is one of the key libraries that RasterFrames builds upon. It provides a Scala language API to working with large raster data with Apache Spark. Ingesting raster data into a Layer is one of the key concepts for creating a dataset for processing on Spark. RasterFrames write data from an appropriate DataFrame into a [GeoTrellis Layer](https://geotrellis.readthedocs.io/en/latest/guide/tile-backends.html). RasterFrames provides a `geotrellis` DataSource that supports both reading and writing of GeoTrellis layers.
 
+> An example is forthcoming.
 
 ## Parquet
 

pyrasterframes/src/main/python/pyrasterframes/__init__.py

@@ -180,18 +180,25 @@ def _geotiff_writer(
     raster_dimensions=None,
     **options):
 
+    def set_dims(parts):
+        parts = [int(p) for p in parts]
+        assert(len(parts) == 2, "Expected dimensions specification to have exactly two components")
+        assert(all([p > 0 for p in parts]), "Expected all components in dimensions to be positive integers")
+        options.update({
+            "imageWidth": parts[0],
+            "imageHeight": parts[1]
+        })
+        try:
+            parts = [int(p) for p in parts]
+            assert(all([p > 0 for p in parts]), 'nice message')
+        except ValueError:
+            raise
+
     if raster_dimensions is not None:
-        if isinstance(raster_dimensions, tuple):
-            options.update({
-                "imageWidth": raster_dimensions[0],
-                "imageHeight": raster_dimensions[1]
-            })
+        if isinstance(raster_dimensions, (list, tuple)):
+            set_dims(raster_dimensions)
         elif isinstance(raster_dimensions, str):
-            parts = raster_dimensions.split(',')
-            options.update({
-                "imageWidth": parts[0],
-                "imageHeight": parts[1]
-            })
+            set_dims(raster_dimensions.split(','))
 
     if crs is not None:
         options.update({
@@ -200,8 +207,11 @@ def _geotiff_writer(
 
     return _aliased_writer(df_writer, "geotiff", path, **options)
 
-# Patch new method on SparkSession to mirror Scala approach
+
+# Patch RasterFrames initialization method on SparkSession to mirror Scala approach
 SparkSession.withRasterFrames = _rf_init
+
+# Patch Kryo serialization initialization method on SparkSession.Builder to mirror Scala approach
 SparkSession.Builder.withKryoSerialization = _kryo_init
 
 # Add the 'asLayer' method to pyspark DataFrame