python - PySpark groupby和最大值选择

我有一个PySpark数据帧

 name   city     date
 satya  Mumbai  13/10/2016
 satya  Pune    02/11/2016
 satya  Mumbai  22/11/2016
 satya  Pune    29/11/2016
 satya  Delhi   30/11/2016
 panda  Delhi   29/11/2016
 brata  BBSR    28/11/2016
 brata  Goa     30/10/2016
 brata  Goa     30/10/2016

我需要找出每个名字最喜欢的城市，逻辑是“如果城市在‘名字’+‘城市’对上出现的城市数量最大，则将城市视为fav_city”。如果发现多个相同的事件，则考虑具有最新日期的城市。将解释：

d = df.groupby('name','city').count()
#name  city  count
brata Goa    2  #clear favourite
brata BBSR   1
panda Delhi  1  #as single so clear favourite
satya Pune   2  ##Confusion
satya Mumbai 2  ##confusion
satya Delhi  1   ##shd be discard as other cities having higher count than this city

#So get cities having max count
dd = d.groupby('name').agg(F.max('count').alias('count'))
ddd = dd.join(d,['name','count'],'left')
#name  count  city
 brata    2   Goa    #fav found
 panda    1   Delhi  #fav found
 satya    2   Mumbai #can't say
 satya    2   Pune   #can't say

如果用户“satya”，我需要返回trx_history并获取具有相同最大计数的城市的最新日期，即：来自“孟买”或“pune”（最后一次交易）（最大日期），将该城市视为fav_city。在这种情况下，“pune”作为“29/11/2016”是最新/最长日期。
但我无法继续如何完成这项工作。
请帮助我的逻辑或如果有更好的解决方案（更快/紧凑的方式），请建议。谢谢。

最佳答案

首次将日期转换为DateType：

df_with_date = df.withColumn(
    "date",
    F.unix_timestamp("date", "dd/MM/yyyy").cast("timestamp").cast("date")
)

下一步groupBy用户和城市，但像这样扩展聚合：

df_agg = (df_with_date
    .groupBy("name", "city")
    .agg(F.count("city").alias("count"), F.max("date").alias("max_date")))

定义窗口：

from pyspark.sql.window import Window

w = Window().partitionBy("name").orderBy(F.desc("count"), F.desc("max_date"))

添加排名：

df_with_rank = (df_agg
    .withColumn("rank", F.dense_rank().over(w)))

和过滤器：

result = df_with_rank.where(F.col("rank") == 1)

您可以使用以下代码检测剩余的重复项：

import sys

final_w = Window().partitionBy("name").rowsBetween(-sys.maxsize, sys.maxsize)
result.withColumn("tie", F.count("*").over(final_w) != 1)

关于python - PySpark groupby和最大值选择，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/40889564/