Эквивалент Pyspark для df.groupby ('id'). Resample ('D'). Last () в пандах

Question

У меня большой стол, как

Я хочу изменить его на новую таблицу: id, date, last_state.

Панды очень легко:

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. Эквивалент Resample в pysaprk является groupby + window:

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

   здесь groupby 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"))
   

   Этот групповой идентификатор и получить последний порядок строк по time_create.

Тем не менее, мне нужен идентификатор groupby, повторная выборка по дням, затем получение последнего порядка строк по time_create.

Я знаю, что эту проблему можно решить, если я использую pandas udf, Применение пользовательских функций к GroupedData в PySpark (с примером работающего python)

Но есть ли способ сделать это только с помощью 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"))