您可以检查此代码：

import pyspark.sql.functions as F
from pyspark.sql import Window

inputData = [
    (1, "AAIC", 11),
    (2, "AAFC", 8),
    (3, "TBIC", 2),
    (4, "TBFC", 1),
    (5, "XYZ", 5),
    (6, "AAIC", 7),
    (7, "JHK", 10),
    (8, "SWE", 3),
    (9, "TBIC", 4),
    (10, "AAFC", 9),
    (11, "AAFC", 6),
]
inputDf = spark.createDataFrame(inputData, schema=["item", "type", "rank"])

preprocessedDf = inputDf.withColumn(
    "type",
    F.when(
        F.col("type").isin(["AAIC", "AAFC", "TBIC", "TBFC"]), F.col("type")
    ).otherwise(F.lit("NON")),
).withColumn(
    "priority",
    F.when(F.col("type") == F.lit("AAIC"), 1).otherwise(
        F.when(F.col("type") == F.lit("AAFC"), 2).otherwise(
            F.when(F.col("type") == F.lit("TBIC"), 3).otherwise(
                F.when(F.col("type") == F.lit("TBFC"), 4).otherwise(F.lit(5))
            )
        )
    ),
)

windowSpec = Window.partitionBy().orderBy("priority", "rank")

preprocessedDf.withColumn("NEW_RANK", F.row_number().over(windowSpec)).drop(
    "priority"
).show()

编码的优先级是硬编码的，如果值比较多，可能会很难维护，如果需要更灵活的话，可以调整这部分
我将所有记录移动到一个分区以计算正确的row_order。这是一个常见问题，很难以分布式方式计算具有给定顺序的一致id。如果您的数据集很大，则可能需要考虑其他问题，可能会更复杂
输出：

+----+----+----+--------+
|item|type|rank|NEW_RANK|
+----+----+----+--------+
|   6|AAIC|   7|       1|
|   1|AAIC|  11|       2|
|  11|AAFC|   6|       3|
|   2|AAFC|   8|       4|
|  10|AAFC|   9|       5|
|   3|TBIC|   2|       6|
|   9|TBIC|   4|       7|
|   4|TBFC|   1|       8|
|   8| NON|   3|       9|
|   5| NON|   5|      10|
|   7| NON|  10|      11|
+----+----+----+--------+