Tweaks to time-series.pymd to lower memory pressure and remove SQL.

Author: metasim

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

@@ -280,6 +280,11 @@ trait RasterFunctions {
       udf(F.rasterize(_: Geometry, _: Geometry, _: Int, cols, rows)).apply(geometry, bounds, value)
     )
 
+  def rf_rasterize(geometry: Column, bounds: Column, value: Column, cols: Column, rows: Column): TypedColumn[Any, Tile] =
+    withTypedAlias("rf_rasterize", geometry)(
+      udf(F.rasterize).apply(geometry, bounds, value, cols, rows)
+    )
+
   /** Reproject a column of geometry from one CRS to another.
     * @param sourceGeom Geometry column to reproject
     * @param srcCRS Native CRS of `sourceGeom` as a literal

pyrasterframes/src/main/python/docs/time-series.pymd

@@ -7,8 +7,8 @@ from IPython.display import display
 import pyrasterframes
 from pyrasterframes.rasterfunctions import *
 import pyrasterframes.rf_ipython
-
-spark = pyrasterframes.get_spark_session()
+# This job is more memory bound, so reduce the concurrent tasks.
+spark = pyrasterframes.get_spark_session("local[4]")
 ```
 
 In this example, we will show how the flexibility of the DataFrame concept for raster data allows a simple and intuitive way to extract a time series from Earth observation data. We will start with our @ref:[built-in MODIS data catalog](raster-catalogs.md#using-built-in-experimental-catalogs).
@@ -35,6 +35,7 @@ park_geo = park_df[park_df.UNIT_CODE=='CUVA'].geometry[0]
 park_geo.wkt[:100]
 ```
 
+
 The entire park boundary is contained in MODIS granule h11 v4. We will simply filter on this granule, rather than using a @ref:[spatial relation](vector-data.md#geomesa-functions-and-spatial-relations). The time period selected should show the change in plant vigor as leaves emerge over the spring and into early summer.
 
 ```python query_catalog
@@ -53,34 +54,30 @@ Now we have a catalog with several months of MODIS data for a single granule. Ho
 
 ```python read_catalog
 raster_cols = ['B01', 'B02',] # red and near-infrared respectively
-park_rf = spark.read.raster(catalog=park_cat.select(['acquisition_date', 'granule_id'] + raster_cols),
-                            catalog_col_names=raster_cols,
-                           ) \
-                    .withColumn('park', st_geomFromWKT(lit(park_geo.wkt)))
-park_rf.createOrReplaceTempView('catalog_read')
+park_rf = spark.read.raster(
+    catalog=park_cat.select(['acquisition_date', 'granule_id'] + raster_cols),
+    catalog_col_names=raster_cols
+    ) \
+    .withColumn('park', st_geomFromWKT(lit(park_geo.wkt)))
 park_rf.printSchema()
+park_rf.persist()                    
 ```
 
 ## Vector and Raster Data Interaction
 
 Now we have the vector representation of the park boundary alongside the _tiles_ of red and near infrared bands. Next, we need to create a _tile_ representation of the park to allow us to limit the time series analysis to pixels within the park. This is similar to the masking operation demonstrated in @ref:[NoData handling](nodata-handling.md#masking).
 
-We do this using two SQL transformations. The first one will reproject the park boundary from coordinates to the MODIS sinusoidal projection. The second one will create a new _tile_ aligned with the imagery containing a value of 1 where the pixels are contained within the park and NoData elsewhere. The code demonstrates that we can use any `rf_` or `st_` function in SQL. For a more thorough demonstration of how Python and SQL codes compare in RasterFrames, see the @ref:[language support section](languages.md).
+We do this using two transformations. The first one will reproject the park boundary from coordinates to the MODIS sinusoidal projection. The second one will create a new _tile_ aligned with the imagery containing a value of 1 where the pixels are contained within the park and NoData elsewhere. 
 
 ```python burn_in
-cr_1 = spark.sql("""
-SELECT *,
-    st_reproject(park, 'EPSG:4326', rf_crs(B01).crsProj4) as park_native
-FROM catalog_read
-""")
-cr_1.createOrReplaceTempView('cr_1')
-
-cr_2 = spark.sql("""
-SELECT *,
-    rf_rasterize(park_native, rf_geometry(B01), 1, rf_dimensions(B01).cols, rf_dimensions(B01).rows) as park_tile
-FROM cr_1
-""").where(rf_tile_sum('park_tile') > 0)
+cr_1 = park_rf.withColumn('park_native', st_reproject('park', lit('EPSG:4326'), rf_crs('B01')))
+
+cr_2 = cr_1 \
+    .withColumn('dims', rf_dimensions('B01')) \
+    .withColumn('park_tile', rf_rasterize('park_native', rf_geometry('B01'), lit(1), 'dims.cols', 'dims.rows')) \
+    .where(rf_tile_sum('park_tile') > 0)
 cr_2.printSchema()
+cr_2.persist()
 ```
 
 ## Create Time Series
@@ -100,7 +97,8 @@ time_series = rf_ndvi \
         .groupby(year('acquisition_date').alias('year'), weekofyear('acquisition_date').alias('week')) \
         .agg(sql_sum('ndvi_wt').alias('ndvi_wt'), sql_sum('wt').alias('wt')) \
         .withColumn('ndvi', col('ndvi_wt') / col('wt'))
-
+time_series.printSchema()
+time_series.persist()
 ```
 
 Finally, we will take a look at the NDVI over time.

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

@@ -55,11 +55,11 @@ def _kryo_init(builder):
     return builder
 
 
-def get_spark_session():
+def get_spark_session(master="local[*]"):
     """ Create a SparkSession with pyrasterframes enabled and configured. """
     from pyrasterframes.utils import create_rf_spark_session
 
-    return create_rf_spark_session()
+    return create_rf_spark_session(master)
 
 
 def _convert_df(df, sp_key=None, metadata=None):

pyrasterframes/src/main/python/pyrasterframes/rasterfunctions.py

@@ -91,11 +91,11 @@ def rf_make_ones_tile(num_cols, num_rows, cell_type=CellType.float64()):
     return Column(jfcn(num_cols, num_rows, _parse_cell_type(cell_type)))
 
 
-def rf_rasterize(geometry_col, bounds_col, value_col, num_cols, num_rows):
+def rf_rasterize(geometry_col, bounds_col, value_col, num_cols_col, num_rows_col):
     """Create a tile where cells in the grid defined by cols, rows, and bounds are filled with the given value."""
     jfcn = RFContext.active().lookup('rf_rasterize')
-    return Column(jfcn(_to_java_column(geometry_col), _to_java_column(bounds_col), _to_java_column(value_col), num_cols,
-                       num_rows))
+    return Column(jfcn(_to_java_column(geometry_col), _to_java_column(bounds_col),
+                       _to_java_column(value_col), _to_java_column(num_cols_col),  _to_java_column(num_rows_col)))
 
 
 def st_reproject(geometry_col, src_crs, dst_crs):

pyrasterframes/src/main/python/pyrasterframes/utils.py

@@ -76,12 +76,12 @@ def find_pyrasterframes_assembly():
     return jarpath[0]
 
 
-def create_rf_spark_session():
+def create_rf_spark_session(master="local[*]"):
     """ Create a SparkSession with pyrasterframes enabled and configured. """
     jar_path = find_pyrasterframes_assembly()
 
     spark = (SparkSession.builder
-             .master("local[*]")
+             .master(master)
              .appName("RasterFrames")
              .config('spark.jars', jar_path)
              .withKryoSerialization()