计算忽略缺失值的百分位数答案

【问题标题】：Calculate percentiles ignoring missing values计算忽略缺失值的百分位数
【发布时间】：2022-03-22 14:28:01
【问题描述】：

我有一个 PySpark 数据框，其中包含 ID 和 BALANCE 列。我正在尝试将balance 列存储到 100 个百分位 (1-100%) 的存储桶中，并计算每个存储桶中有多少个 ID。

我不能使用任何与 RDD 相关的东西，我只能使用 PySpark 语法。我试过下面的代码。

w = Window.orderBy(df.BALANCE)
test = df.withColumn('percentile_col', F.percent_rank().over(w))

我希望得到一个新列，它会自动计算 BALANCE 列中每个数据点的百分位数并忽略缺失值。

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql null percentile

【解决方案1】：

测试数据：

from pyspark.sql import SparkSession, functions as F, Window
spark = SparkSession.builder.getOrCreate()
df = spark.range(12).withColumn(
    'balance',
    F.when(~F.col('id').isin([0, 1, 2, 3, 4]), F.col('id') + 500))
df.show()
#+---+-------+
#| id|balance|
#+---+-------+
#|  0|   null|
#|  1|   null|
#|  2|   null|
#|  3|   null|
#|  4|   null|
#|  5|    505|
#|  6|    506|
#|  7|    507|
#|  8|    508|
#|  9|    509|
#| 10|    510|
#| 11|    511|
#+---+-------+

percent_rank 会给你精确的百分位数 - 结果可能有很长的十进制值。这就是为什么单独percent_rank 可能无法提供您想要的。

null_value = F.col('balance').isNull()
df = (
    df.filter(~null_value)
    .withColumn('percentile', F.percent_rank().over(Window.orderBy('balance')))
    .unionByName(df.filter(null_value), True)
)
df.show()
#+---+-------+-------------------+
#| id|balance|         percentile|
#+---+-------+-------------------+
#|  5|    505|                0.0|
#|  6|    506|0.16666666666666666|
#|  7|    507| 0.3333333333333333|
#|  8|    508|                0.5|
#|  9|    509| 0.6666666666666666|
#| 10|    510| 0.8333333333333334|
#| 11|    511|                1.0|
#|  0|   null|               null|
#|  1|   null|               null|
#|  2|   null|               null|
#|  3|   null|               null|
#|  4|   null|               null|
#+---+-------+-------------------+

以下应该有效。添加了舍入步骤。

null_value = F.col('balance').isNull()
pr = F.percent_rank().over(Window.orderBy('balance'))
df = (
    df.filter(~null_value)
    .withColumn('bucket', F.when(pr == 0, 1).otherwise(F.ceil(pr * 100)))
    .unionByName(df.filter(null_value), True)
)
df.show()
#+---+-------+------+
#| id|balance|bucket|
#+---+-------+------+
#|  5|    505|     1|
#|  6|    506|    17|
#|  7|    507|    34|
#|  8|    508|    50|
#|  9|    509|    67|
#| 10|    510|    84|
#| 11|    511|   100|
#|  0|   null|  null|
#|  1|   null|  null|
#|  2|   null|  null|
#|  3|   null|  null|
#|  4|   null|  null|
#+---+-------+------+

您也可以考虑ntile。每个值都被添加到 n 个“桶”之一中。

当 n=100 时（测试表少于 100 项，因此只有第一个“桶”获取值）：

null_value = F.col('balance').isNull()
df = (
    df.filter(~null_value)
    .withColumn('ntile_100', F.ntile(100).over(Window.orderBy('balance')))
    .unionByName(df.filter(null_value), True)
)
df.show()
#+---+-------+---------+
#| id|balance|ntile_100|
#+---+-------+---------+
#|  5|    505|        1|
#|  6|    506|        2|
#|  7|    507|        3|
#|  8|    508|        4|
#|  9|    509|        5|
#| 10|    510|        6|
#| 11|    511|        7|
#|  0|   null|     null|
#|  1|   null|     null|
#|  2|   null|     null|
#|  3|   null|     null|
#|  4|   null|     null|
#+---+-------+---------+

当 n=4 时：

null_value = F.col('balance').isNull()
df = (
    df.filter(~null_value)
    .withColumn('ntile_100', F.ntile(4).over(Window.orderBy('balance')))
    .unionByName(df.filter(null_value), True)
)
df.show()
#+---+-------+---------+
#| id|balance|ntile_100|
#+---+-------+---------+
#|  5|    505|        1|
#|  6|    506|        1|
#|  7|    507|        2|
#|  8|    508|        2|
#|  9|    509|        3|
#| 10|    510|        3|
#| 11|    511|        4|
#|  0|   null|     null|
#|  1|   null|     null|
#|  2|   null|     null|
#|  3|   null|     null|
#|  4|   null|     null|
#+---+-------+---------+

【讨论】：

【解决方案2】：

试试这个。我们首先检查 df.Balance 列是否有 Null 值。如果它有 Null 值，我们将显示 None。否则将应用 percent_rank() 函数。

from pyspark.sql import functions as F

w = Window.orderBy(df.BALANCE)

test = df.withColumn('percentile_col',when(df.BALANCE.isNull(), lit(None)).otherwise(F.percent_rank().over(w)))

【讨论】：

不幸的是，这并没有忽略空值。这只是隐藏了 BALANCE 为空的条目的 percent_rank 结果。