Let's try the following :

from pyspark.ml.image import ImageSchema, ImageTransformer
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
 
angle = 90
 
def rotate_image(image):
    np_array = ImageSchema.toNDArray(image)
    rotated = np.rot90(np_array, k=angle//90, axes=(1, 0))
    return ImageSchema.fromNDArray(rotated)
 
rotate_udf = udf(rotate_image, StringType())
 
image_df = spark.read.format("image").load("path/to/image/directory")
rotated_images_df = ImageTransformer().setOutputCol("rotated").transform(image_df).withColumn("rotated", rotate_udf("image"))