您可以使用窗口在每行关联其类别中存在的最大N。然后将条件应用于此新列以筛选类别。

w = Window.partitionBy("Cod", "Category")

df = df.withColumn("max_N_in_category", F.max("N").over(w))

N = 3
df = df \
    .filter(F.col("max_N_in_category") > N) \
    .drop("max_N_in_category")

数据类型

df =spark.createDataFrame([(1 ,   'A',         1 ),
(1  ,  'A' ,        2),
(1   , 'A'  ,       3),
(1    ,'B' ,        1),
(1  ,  'B' ,        2),
(1   , 'B' ,        3),
(1   , 'B' ,        4),
(1   , 'B' ,        5),
(2   , 'D' ,        1),
(3   , 'Z' ,        1),
(3   , 'Z'  ,       2)],
('Cod',  'Category',  'N'))



new = (df.withColumn('check', collect_list('N').over(Window.partitionBy('cod','Category')))#create array per group and keep in column check
       
       .where(expr("exists(check, c -> array_contains(check,3))"))#Filter arrays that do not contaiin 3
       .drop('check')#drop column check
      ).show()

结果

+---+--------+---+
|Cod|Category|  N|
+---+--------+---+
|  1|       A|  1|
|  1|       A|  2|
|  1|       A|  3|
|  1|       B|  1|
|  1|       B|  2|
|  1|       B|  3|
|  1|       B|  4|
|  1|       B|  5|

如果不想使用窗口函数，那么可以通过groupBy和filter with isin()来完成：

df.filter(df.Category.isin([x.Category for x in df.groupBy("Category").max("N").collect() if x["max(N)"] > 3]))

[Out]:
+---+--------+---+
|Cod|Category|  N|
+---+--------+---+
|  1|       B|  1|
|  1|       B|  2|
|  1|       B|  3|
|  1|       B|  4|
|  1|       B|  5|
|  3|       Z|  1|
|  3|       Z|  2|
|  3|       Z|  3|
|  3|       Z|  4|
+---+--------+---+