How to adapt pandas diff() to pyspark?

user_12242625 0
Viewed 41

In Pandas I can ran df_raw_node["delta"] = df_raw_node.sort_values(["date", "hour", "node"])["snr_avg_up"].diff(1)for getting a new column with previous value of this group.

I tried the same with pyspark.

df_spark_joined = (df_spark.join(F.broadcast(df_spark_cm2plot), 
                      df_spark.cm_mac==df_spark_cm2plot.cpe, 
                      "right")
.where(F.col("cmts_id").rlike(stations))
.sort(["date", "hour"])               
.groupby(["date", "hour", "node"])
.diff(1)

But here I'm getting an error 'GroupedData' object has no attribute 'diff'.

Is there another way to do it easily, as in pandas?

As it is spark on a BigData cluster I can't provide a MWE and am therefore grateful for any idea and do not expect a fully coded solution

pyspark python
1 Answers

Pandas' diff() subtracts the current value and the previous value of column in a sorted dataframe. This can be achieved by using lag window function and subtracting the current value.

data_sdf. \
    withColumn('diff_with_prevval', 
               func.col('val') - func.lag('val').over(wd.partitionBy().orderBy('idx'))
               )

Add the grouping columns in partitionBy() as per requirement. There should at least be a sorting field for the lag to work correctly, as spark does not retain a dataframe's sort.

user_82795850
Related