你可以做一个左半连接 user_id 至少有 rank >=6 :

df1 = df.join(
    df.filter("rank >= 6").select("user_id").distinct(),
    ["user_id"],
    "left_semi"
)

df1.show()

# +-------+------+---------+----+

# |user_id|  item| category|rank|

# +-------+------+---------+----+

# | user_2| item6|categoryB|   1|

# | user_2| item7|categoryB|   2|

# | user_2| item8|categoryB|   3|

# | user_2| item9|categoryA|   4|

# | user_2|item10|categoryE|   5|

# | user_2|item11|categoryE|   6|

# | user_3| item6|categoryB|   1|

# | user_3| item7|categoryB|   2|

# | user_3| item8|categoryB|   3|

# | user_3| item9|categoryA|   4|

# | user_3|item10|categoryE|   5|

# | user_3|item11|categoryE|   6|

# +-------+------+---------+----+

或者 max_rank >= 6 窗口上方：

from pyspark.sql import functions as F, Window

df1 = df.withColumn("max_rn", F.max("rank").over(Window.partitionBy("user_id")))\
    .filter("max_rn >= 6").drop("max_rn")

您可以使用每个用户的计数进行筛选：

from pyspark.sql import functions as F, Window

df2 = df.withColumn(
    'cnt', 
    F.count('*').over(Window.partitionBy('user_id'))
).filter('cnt >= 6')

df2.show()
+-------+------+---------+----+---+
|user_id|  item| category|rank|cnt|
+-------+------+---------+----+---+
| user_2| item6|categoryB|   1|  6|
| user_2| item7|categoryB|   2|  6|
| user_2| item8|categoryB|   3|  6|
| user_2| item9|categoryA|   4|  6|
| user_2|item10|categoryE|   5|  6|
| user_2|item11|categoryE|   6|  6|
| user_3| item6|categoryB|   1|  6|
| user_3| item7|categoryB|   2|  6|
| user_3| item8|categoryB|   3|  6|
| user_3| item9|categoryA|   4|  6|
| user_3|item10|categoryE|   5|  6|
| user_3|item11|categoryE|   6|  6|
+-------+------+---------+----+---+