您可以使用Window，并按id_prd和type进行分区，并按price排序，然后仅在有type = AA的情况下获得row_number

from pyspark.sql.functions import row_number, desc, col, lit, when
from pyspark.sql.window import Window

data = [
    (1, "A", "SS", 10),
    (2, "A", "AA", 20),
    (3, "A", "AA", 25),
    (1, "B", "AA", 55),
    (2, "B", "SS", 50),
    (3, "B", "AA", 75),
    (4, "B", "AA", 80)]

df = spark.createDataFrame(data, ["id_cnt", "id_prd", "type", "price"])
            
window = Window.partitionBy("id_prd", "type").orderBy(desc("price"))

df.withColumn("row_number", when(col("type") == lit("AA"), row_number().over(window)).otherwise(None)).show()

+------+------+----+-----+----------+                                           
|id_cnt|id_prd|type|price|row_number|
+------+------+----+-----+----------+
|     3|     A|  AA|   25|         1|
|     2|     A|  AA|   20|         2|
|     1|     A|  SS|   10|      null|
|     4|     B|  AA|   80|         1|
|     3|     B|  AA|   75|         2|
|     1|     B|  AA|   55|         3|
|     2|     B|  SS|   50|      null|
+------+------+----+-----+----------+