PySpark中计算各Key对应平均值的百分位数（模拟PERCENTRANK.INC）

核心逻辑对齐 #

Excel的PERCENTRANK.INC(array, x)计算逻辑为：

对于分组内的目标值x（即每个Key对应的avg），百分位数 = (小于x的元素数量 + 0.5 * 等于x的元素数量) / (分组内总元素数 - 1)
若分组内只有1个元素，直接返回0

实现步骤（自动化处理多Key） #

使用PySpark窗口函数按Key分组，完成以下操作：

1. 计算每个分组的总元素数
2. 统计每个分组中小于avg的values数量
3. 统计每个分组中等于avg的values数量
4. 代入公式计算百分位数

完整代码示例 #

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

# 初始化SparkSession
spark = SparkSession.builder.appName("PercentRankInc").getOrCreate()

# 示例数据（替换为你的实际DataFrame）
data = [
    (10, "A", 15),
    (15, "A", 15),
    (20, "A", 15),
    (5, "B", 7),
    (7, "B", 7),
    (9, "B", 7),
    (3, "C", 3)
]
df = spark.createDataFrame(data, ["values", "Key", "avg"])

# 定义分组窗口
group_window = Window.partitionBy("Key")

# 计算分组内总元素数
df_with_count = df.withColumn("total_elements", F.count("values").over(group_window))

# 统计小于avg的元素数量
df_with_less = df_with_count.withColumn(
    "count_less",
    F.sum(F.when(F.col("values") < F.col("avg"), 1).otherwise(0)).over(group_window)
)

# 统计等于avg的元素数量
df_with_equal = df_with_less.withColumn(
    "count_equal",
    F.sum(F.when(F.col("values") == F.col("avg"), 1).otherwise(0)).over(group_window)
)

# 计算PERCENTRANK.INC值
final_df = df_with_equal.withColumn(
    "percent_rank_inc",
    F.when(
        F.col("total_elements") == 1,
        0.0
    ).otherwise(
        (F.col("count_less") + 0.5 * F.col("count_equal")) / (F.col("total_elements") - 1)
    )
).drop("total_elements", "count_less", "count_equal")

# 展示结果
final_df.show()

结果说明 #

运行上述代码后，每个Key对应的avg值会得到和ExcelPERCENTRANK.INC完全一致的百分位数结果，且自动处理所有分组的Key，无需手动逐个处理。

内容的提问来源于stack exchange，提问作者Mohammad Tayyab