多列的 PySpark 百分位数

Question

我想使用 PySpark 将 PySpark 数据帧的多个数字列转换为其百分位值，而不更改其顺序。

例如给定一个列名数组arr = [Salary, Age, Bonus]，将列转换为百分位数。

输入

+----------+-------------+---------+--------+-----+-------+
| Empl. No | Dept        | Pincode | Salary | Age | Bonus |
+----------+-------------+---------+--------+-----+-------+
| 1        | HR          | 111     | 1000   | 45  | 100   |
| 2        | Sales       | 596     | 500    | 30  | 50    |
| 3        | Manufacture | 895     | 600    | 50  | 400   |
| 4        | HR          | 212     | 700    | 26  | 60    |
| 5        | Business    | 754     | 350    | 18  | 22    |
+----------+-------------+---------+--------+-----+-------+

预期输出

+----------+-------------+---------+--------+-----+-------+
| Empl. No | Dept        | Pincode | Salary | Age | Bonus |
+----------+-------------+---------+--------+-----+-------+
| 1        | HR          | 111     | 100    | 80  | 80    |
| 2        | Sales       | 596     | 40     | 60  | 40    |
| 3        | Manufacture | 895     | 60     | 100 | 100   |
| 4        | HR          | 212     | 80     | 40  | 60    |
| 5        | Business    | 754     | 20     | 20  | 20    |
+----------+-------------+---------+--------+-----+-------+

列表中给定元素“x”的百分位数公式 =（小于“x”的元素数/总元素数）*100。

Answer 1

您可以将percentile_approx 用于此，将groupBy 与您想要计算百分位数的所需列结合使用。

内置 Spark > 3.x

input_list = [
    (1,"HR",111,1000,45,100)
    ,(2,"Sales",112,500,30,50)
    ,(3,"Manufacture",127,600,50,500)
    ,(4,"Hr",821,700,26,60)
    ,(5,"Business",754,350,18,22)
]

sparkDF = sql.createDataFrame(input_list,['emp_no','dept','pincode','salary','age','bonus'])

sparkDF.groupBy(['emp_no','dept']).agg(
    *[ F.first(F.col('pincode')).alias('pincode') ]
    ,*[ F.percentile_approx(F.col(col),0.95).alias(col) for col in ['salary','age','bonus'] ]
).show()

+------+-----------+-------+------+---+-----+
|emp_no|       dept|pincode|salary|age|bonus|
+------+-----------+-------+------+---+-----+
|     3|Manufacture|    127|   600| 50|  500|
|     1|         HR|    111|  1000| 45|  100|
|     2|      Sales|    112|   500| 30|   50|
|     5|   Business|    754|   350| 18|   22|
|     4|         Hr|    821|   700| 26|   60|
+------+-----------+-------+------+---+-----+

Answer 2

Spark 有一个用于计算百分位数的窗口函数，称为 percent_rank。

测试df：

from pyspark.sql import SparkSession, functions as F, Window as W
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame(
    [(1, "HR", 111, 1000, 45, 100),
     (2, "Sales", 596, 500, 30, 50),
     (3, "Manufacture", 895, 600, 50, 400),
     (4, "HR", 212, 700, 26, 60),
     (5, "Business", 754, 350, 18, 22)],
    ['Empl_No', 'Dept', 'Pincode', 'Salary', 'Age', 'Bonus'])
df.show()
#  +-------+-----------+-------+------+---+-----+
#  |Empl_No|       Dept|Pincode|Salary|Age|Bonus|
#  +-------+-----------+-------+------+---+-----+
#  |      1|         HR|    111|  1000| 45|  100|
#  |      2|      Sales|    596|   500| 30|   50|
#  |      3|Manufacture|    895|   600| 50|  400|
#  |      4|         HR|    212|   700| 26|   60|
#  |      5|   Business|    754|   350| 18|   22|
#  +-------+-----------+-------+------+---+-----+

percent_rank 的工作方式是最小值为百分位数 0，最大值为 1。

arr = ['Salary', 'Age', 'Bonus']
df = df.select(
    *[c for c in df.columns if c not in arr],
    *[F.percent_rank().over(W.orderBy(c)).alias(c) for c in arr]
).sort('Empl_No')

df.show()
#  +-------+-----------+-------+------+----+-----+
#  |Empl_No|       Dept|Pincode|Salary| Age|Bonus|
#  +-------+-----------+-------+------+----+-----+
#  |      1|         HR|    111|   1.0|0.75| 0.75|
#  |      2|      Sales|    596|  0.25| 0.5| 0.25|
#  |      3|Manufacture|    895|   0.5| 1.0|  1.0|
#  |      4|         HR|    212|  0.75|0.25|  0.5|
#  |      5|   Business|    754|   0.0| 0.0|  0.0|
#  +-------+-----------+-------+------+----+-----+

但是，您的期望有些不同。您希望它假定 0 作为最小值，即使它不存在于列中。

为了解决这个问题，我将添加一个值为 0 的行，然后将其删除。

arr = ['Salary', 'Age', 'Bonus']

# Adding a row containing 0 values
df = df.limit(1).withColumn('Dept', F.lit('_tmp')).select(
    *[c for c in df.columns if c not in arr],
    *[F.lit(0).alias(c) for c in arr]
).union(df)

# Calculating percentiles
df = df.select(
    *[c for c in df.columns if c not in arr],
    *[F.percent_rank().over(W.orderBy(c)).alias(c) for c in arr]
).sort('Empl_No')

# Removing the fake row
df = df.filter("Dept != '_tmp'")

df.show()
#  +-------+-----------+-------+------+---+-----+
#  |Empl_No|       Dept|Pincode|Salary|Age|Bonus|
#  +-------+-----------+-------+------+---+-----+
#  |      1|         HR|    111|   1.0|0.8|  0.8|
#  |      2|      Sales|    596|   0.4|0.6|  0.4|
#  |      3|Manufacture|    895|   0.6|1.0|  1.0|
#  |      4|         HR|    212|   0.8|0.4|  0.6|
#  |      5|   Business|    754|   0.2|0.2|  0.2|
#  +-------+-----------+-------+------+---+-----+

如果您愿意，可以将百分位数乘以 100：

*[(100 * F.percent_rank().over(W.orderBy(c))).alias(c) for c in arr]

然后你得到...

+-------+-----------+-------+------+-----+-----+
|Empl_No|       Dept|Pincode|Salary|  Age|Bonus|
+-------+-----------+-------+------+-----+-----+
|      1|         HR|    111| 100.0| 80.0| 80.0|
|      2|      Sales|    596|  40.0| 60.0| 40.0|
|      3|Manufacture|    895|  60.0|100.0|100.0|
|      4|         HR|    212|  80.0| 40.0| 60.0|
|      5|   Business|    754|  20.0| 20.0| 20.0|
+-------+-----------+-------+------+-----+-----+