[SEDONA-406] Raster deserializer for PySpark

[Kontinuation]

Did you read the Contributor Guide?

• Yes, I have read Contributor Rules and Contributor Development Guide

Is this PR related to a JIRA ticket?

• Yes, the URL of the associated JIRA ticket is https://issues.apache.org/jira/browse/SEDONA-406. The PR name follows the format [SEDONA-XXX] my subject.

What changes were proposed in this PR?

API changes

This PR adds a new class SedonaRaster to sedona python package. Raster objects in sedona will be converted to SedonaRaster objects in python when collecting raster objects in PySpark:

rows = df_rast.collect()
rast = rows[0]['rast']  # rast is a SedonaRaster object

# You can get the metadata of raster by accessing the properties of SedonaRaster objects
print(rast.width, rast.height)
print(rast.affine_trans)
print(rast.crs_wkt)

# You can get the band data as numpy array
arr = rast.as_numpy()

# You can also get a rasterio DatasetReader object
ds = rast.as_rasterio()

# Please close the SedonaRaster after using it to free up resources allocated for the rasterio DatasetReader object
rast.close()

Users can define PandasUDFs taking raster object as parameter. Please use the deserialize function in sedona.raster.raster_serde module to deserialize the bytes to SedonaRaster object before processing it. Please note that this only works with Spark >= 3.4.0.

# A Python Pandas UDF that takes a geometry as input
@pandas_udf(IntegerType())
def pandas_udf_raster_as_param(s: pd.Series) -> pd.Series:
    from sedona.raster import raster_serde

    def func(x):
        with raster_serde.deserialize(x) as raster:
            arr = raster.as_numpy()
            return int(np.sum(arr))

    return s.apply(func)

spark.udf.register("pandas_udf_raster_as_param", pandas_udf_raster_as_param)

Internal changes

• Changed the serialization format of RasterUDT to a language-neutral format
  • Notably, CRS is now serialized to WKT instead of using the Java serializer.
  • This also significantly improved the performance of raster serialization/deserialization, since the new Kryo serailizer is way faster than Java serializer we used before.
• Added a raster deserializer for PySpark

How was this patch tested?

Added new tests

Did this PR include necessary documentation updates?

• Yes. I've updated the documentation.

[Kontinuation]

The R test failure is unrelated to this PR. The recent updates of sparklyr or dbplyr caused this problem. See sparklyr/sparklyr#3429.

[jiayuasu]

Should we put these jars in geotools-wrapper?

[Kontinuation]

These jars are already in geotools-wrapper, so we can instead put the geotools-wrapper jar to the $SPARK_HOME/jars/ folder, and build spark-shaded jar without -Dgeotools. However, this won't be able to test dependency changes such as adding jiffle as a new dependency.

I can update the document to use geotools-wrapper instead of directly using JAI jars, since it is much easier (no need to rebuild with -Dgeotools for testing sedona python), and covers most cases.

[jiayuasu]

This is fine then. We don't need to update this.

[jiayuasu]

Can you add one more section to explain how to write a regular Python User Defined Function (not Pandas UDF) to work on the raster type? I understand that the UDF cannot return a raster type directly since we only have a Python deserializer, but with the RS_MakeRaster() + NumPy array, we can still construct the raster type. It is important to show this workflow. Maybe we can show this in a separate Doc PR?

[Kontinuation]

New section added.