[GitHub] [spark] adsk2050 commented on pull request #21082: [SPARK-22239][SQL][Python] Enable grouped aggregate pandas UDFs as window functions with unbounded window frames

[GitBox <gi...@apache.org>]

adsk2050 commented on PR #21082:
URL: https://github.com/apache/spark/pull/21082#issuecomment-1163170533

   Hello! this is great work! Thank you for contributing. This code will enable to run functions on window, which take in pd.Series -> Any.
   
   I am wondering if GROUPED_MAP pandas UDF as window functions is also in pipeline or not? 
   (Basically pd.Series -> pd.Series over Window.) 
   For example:
   ```
   from pyspark.sql import functions as F
   from pyspark.sql.types import *
   
   def doCoolStuff(df: pd.DataFrame) -> pd.DataFrame:
     events = df["event"].to_list()
     count = 1
     sets = []
     for event in events:
       sets.append(str(count))
       if event=="buy":
         count+=1   
     df["coolStuff"] = pd.Series(data=sets)
     return df
   
   df = spark.createDataFrame(pd.DataFrame([[1, random.choice(list(range(10))), i, random.random()] for i in range(100)], columns=["user_id", "source_id", "epoch_timestamp", "event_prob"]))\
   .withColumn("event", F.when(F.col("event_prob")>F.lit(0.9), "buy").otherwise("view"))\
   .withColumn("coolStuff", F.lit(""))\
   .persist()
   
   doCoolStuffPDUDF = F.pandas_udf(
     f=doCoolStuff,
     returnType=df.schema,
     functionType=F.PandasUDFType.GROUPED_MAP)
   
   df\
   .orderBy(F.col("epoch_timestamp"))\
   .groupby("user_id", "source_id")\
   .apply(doCoolStuffPDUDF)\
   .orderBy(F.col("user_id"), F.col("source_id"), F.col("epoch_timestamp").desc())\
   .display()
   ```
   
   This could simplified to:
   
   ```
   from pyspark.sql import functions as F
   from pyspark.sql.types import *
   from pyspark.sql.window import Window
   
   def doCoolStuff(events: pd.Series) -> pd.Series:
     count = 1
     sets = []
     for event in events:
       sets.append(str(count))
       if event=="buy":
         count+=1   
     return pd.Series(data=sets)
   
   doCoolStuffPDUDF = F.pandas_udf(
     f=doCoolStuff,
     returnType=StringType(),
     functionType=F.PandasUDFType.GROUPED_MAP)
   
   df = spark.createDataFrame(pd.DataFrame([[1, random.choice(list(range(10))), i, random.random()] for i in range(100)], columns=["user_id", "source_id", "epoch_timestamp", "event_prob"]))\
   .withColumn("event", F.when(F.col("event_prob")>F.lit(0.9), "buy").otherwise("view"))\
   .withColumn("coolStuff", doCoolStuffPDUDF(F.col("event"))\
                                           .over(Window.partitionBy("user_id", "source_id").orderBy(F.col("epoch_timestamp"))\
   .orderBy(F.col("user_id"), F.col("source_id"), F.col("epoch_timestamp").desc())\
   .persist()
   
   df.display()
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org