pandas - Pyspark 相当于 pandas 中的 df.groupby ('id' ).resample ('D' ).last()

Question

我有一张像这样的大 table

我想将其更改为新表: id、date、last_state 。

pandas 非常容易:

df['time_create'] = pd.to_datetime(df['time_create'])
df = df.set_index('time_create')
df = df.sort_index()
df = df.groupby('id').resample('D').last().reset_index()

但是pyspark实现起来非常困难。

我知道:

1. pysaprk 中的重采样等效项是 groupby + window :

   grouped = df.groupBy('store_product_id', window("time_create", "1 day")).agg(sum("Production").alias('Sum Production'))
   

   此处按 store_product_id 分组，按天重新采样并计算总和

2. 分组并查找第一个或最后一个:

   引用https://stackoverflow.com/a/35226857/1637673

   w = Window().partitionBy("store_product_id").orderBy(col("time_create").desc())
   (df
     .withColumn("rn", row_number().over(w))
     .where(col("rn") == 1)
     .select("store_product_id", "time_create", "state"))
   

   这个groupby id并通过time_create获取最后一行的顺序。

但是我需要的是按 id 分组，按天重新采样，然后按 time_create 获取最后一行顺序。

我知道如果我使用 pandas udf 可以解决这个问题，Applying UDFs on GroupedData in PySpark (with functioning python example)

但是有什么方法可以仅通过 pyspark 来做到这一点吗？

Answer 1

只需partitionBy("store_product_id", "date") 即可解决问题

w = Window().partitionBy("store_product_id", "date").orderBy(col("time_create").desc())
x = (df
    .withColumn("rn", row_number().over(w))
    .where(col("rn") == 1)
    .select("store_product_id", "time_create", "state"))