我有一个可能工作的解决方案，但我不确定它是否在时间和资源方面的好解决方案，所以如果有人知道如何改进它，请留下评论。目前我无法找出其他任何东西，但也许它会对你有用。我有一种感觉，有一些技巧，我不知道做得更聪明：D

import datetime 
import pyspark.sql.functions as F

x = [(123,"no", datetime.date(2020,10,25)),
    (123,"yes", datetime.date(2020,10,22)),
    (4141,"no", datetime.date(2020,10,25)),
    (4141,"no", datetime.date(2020,10,22)),
    (4141,"no", datetime.date(2020,10,15)),
    (5555,"yes", datetime.date(2020,10,25)),
    (5555,"no", datetime.date(2020,10,22)),
    (5555,"no", datetime.date(2020,10,15))]
df = spark.createDataFrame(x, schema=['customer_id', 'status', 'date'])
groupedDf = df.groupBy(F.col('customer_id'), F.col('status')).agg(F.max("date").alias("most_recent_date")).cache()
trueDf = groupedDf.filter(F.col('status') == F.lit('yes'))
falseDf = groupedDf.filter(F.col('status') == F.lit('no'))
falseWithNoCorrecpondingTrueDf = falseDf.join(trueDf, falseDf.customer_id == trueDf.customer_id, "anti")
finalDf = falseWithNoCorrecpondingTrueDf.union(trueDf)

dfs不需要单独的变量，我添加它是为了使它更具描述性。
分步说明：

• 首先，我对记录进行分组，以获取customer_id和status的最大日期
• 然后缓存分组的结果，因为我知道它将被使用两次，我不想计算它两次
• 我将group by的结果拆分为两部分，一部分为“是”，另一部分为“否”。
• 我放弃了对应于“是”的“否”，因为根据您的逻辑，它们不会被使用
• 我正在做一个“no”的并集，剩下的都是“yes”，这应该会给予我你想要的结果df

示例作业的输出：

+-----------+------+----------------+
|customer_id|status|most_recent_date|
+-----------+------+----------------+
|       4141|    no|      2020-10-25|
|        123|   yes|      2020-10-22|
|       5555|   yes|      2020-10-25|
+-----------+------+----------------+