получить строку, соответствующую последней отметке времени в pyspark

Question

У меня есть датафрейм как:

+--------------+-----------------+-------------------+
|          ecid|    creation_user| creation_timestamp|
+--------------+-----------------+-------------------+
|ECID-195000300|USER_ID1          |2018-08-31 20:00:00|
|ECID-195000300|USER_ID2          |2016-08-31 20:00:00|

Мне нужно иметь строку с самой ранней меткой времени как:

+--------------+-----------------+-------------------+
    |          ecid|    creation_user| creation_timestamp|
    +--------------+-----------------+-------------------+
    |ECID-195000300|USER_ID2          |2016-08-31 20:00:00|

Как я могу получить это в pyspark: Я пытался

 df.groupBy("ecid").agg(min("creation_timestamp"))

Однако я только получаю поле ecid и timestamp. Я хочу, чтобы все поля, а не только два поля

Answer 1

Использование окна row_number функция с разделением на на ecid и , заказ на на creation_timestamp.

Example:

#sampledata

df=spark.createDataFrame([("ECID-195000300","USER_ID1","2018-08-31 20:00:00"),("ECID-195000300","USER_ID2","2016-08-31 20:00:00")],["ecid","creation_user","creation_timestamp"])

from pyspark.sql import Window
from pyspark.sql.functions import *

w = Window.partitionBy('ecid').orderBy("creation_timestamp")

df.withColumn("rn",row_number().over(w)).filter(col("rn") ==1).drop("rn").show()
#+--------------+-------------+-------------------+
#|          ecid|creation_user| creation_timestamp|
#+--------------+-------------+-------------------+
#|ECID-195000300|     USER_ID2|2016-08-31 20:00:00|
#+--------------+-------------+-------------------+

Answer 2

Я думаю, вам понадобится функция window + фильтр для этого. Я могу предложить вам следующее непроверенное решение:

import pyspark.sql.window as psw
import pyspark.sql.functions as psf

w = psw.Window.partitionBy('ecid')
df = (df.withColumn("min_tmp", psf.min('creation_timestamp').over(w))
         .filter(psf.col("min_tmp") == psf.col("creation_timestamp")) 
)

Функция window позволяет вам возвращать min для каждого ecid как новый столбец вашего DataFrame