在pyspark >= 3.3.0中使用MAX_BY

from pyspark.sql.window import Window
from pyspark.sql import functions as F

w = (
    Window
    .partitionBy('category')
    .rangeBetween(Window.unboundedPreceding, Window.unboundedFollowing)
)
mdt = F.max_by('datetime', 'value').over(w)
df2 = df.withColumn('datetime_max', mdt)

字符串

from pyspark.sql.functions import max

df = df.join(
    df.groupby('category'). \
    agg(max('value').alias('datetimeMax')), \
    on=['category'])
df.show()

字符串

在这种情况下使用window函数（row_number,max），通过在category上定义分区by，并在value上降序排序by。

• when(row_number().over(w) == 1,->当row_number=1时，获取datetime值，否则保留为null
• max(when(row_number().over(w) == 1,col("datetime")))->获取窗口的最大值并填充所有行
  Example:

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

w = Window.partitionBy('category').orderBy(desc('value'))

w = Window.partitionBy('category').orderBy(desc('value'))

df.withColumn("datetimeMax",max(when(row_number().over(w) == 1,col("datetime"))).over(w)).show(100,False)

#+--------+--------+-----+-----------+
#|category|datetime|value|datetimeMax|
#+--------+--------+-----+-----------+
#|a       |date4   |50   |date4      |
#|a       |date2   |30   |date4      |
#|a       |date5   |30   |date4      |
#|a       |date3   |20   |date4      |
#|a       |date1   |10   |date4      |
#+--------+--------+-----+-----------+

字符串