Convert Oracle PL/SQL to PySpark and Spark SQL on Databricks

-- Procedure: process_customer_segments
CREATE OR REPLACE PROCEDURE process_segments AS
  TYPE t_cust IS TABLE OF customers%ROWTYPE;
  l_custs t_cust;
  CURSOR c_active IS
    SELECT * FROM customers
    WHERE status = 'ACTIVE'
      AND total_spend > 1000;
BEGIN
  OPEN c_active;
  LOOP
    FETCH c_active BULK COLLECT
      INTO l_custs LIMIT 5000;
    EXIT WHEN l_custs.COUNT = 0;

    FORALL i IN 1..l_custs.COUNT
      INSERT INTO customer_segments VALUES (
        l_custs(i).cust_id,
        DECODE(l_custs(i).total_spend,
          NULL, 'Unknown',
          'Platinum'),
        NVL(l_custs(i).region, 'Unassigned'),
        SYSDATE
      );
    COMMIT;
  END LOOP;
  CLOSE c_active;
END;

# PL/SQL procedure → PySpark on Databricks
from pyspark.sql import functions as F

def process_segments():
    df = spark.read.table("customers")

    segmented = (
        df.filter(
            (F.col("status") == "ACTIVE") &
            (F.col("total_spend") > 1000)
        )
        .withColumn("segment",
            F.when(F.col("total_spend").isNull(),
                   "Unknown")
             .otherwise("Platinum"))
        .withColumn("region",
            F.coalesce(F.col("region"),
                       F.lit("Unassigned")))
        .withColumn("processed_date",
            F.current_date())
        .select("cust_id", "segment",
                "region", "processed_date")
    )

    segmented.write.format("delta") \
        .mode("append") \
        .saveAsTable("customer_segments")

process_segments()