如何用pyspark中的平均值替换异常值？

Question

我想知道如何用平均值替换异常值。我有数据框，我可以找到异常值并过滤行，现在我想用平均值替换它。我该怎么做？

df 是这样的：

       a     b
1      27    0
2      10    1
3      80    2
4      21    3
5      46    4
6      100   5

找到 IQR 后，我得到了异常值：

Upper = 75
lower = 12
outliers = df.filter((df['a'] > upper) | (df['a'] < lower))
2      10    1
3      80    2
6      100   5

现在我找到了平均值：

from pyspark.sql.functions import mean as _mean, col
mean= df.select(_mean(col('a')).alias('mean')).collect()
mean = mean[0]['mean']
mean : 31.333

现在我不明白如何通过将平均值舍入为 31 并将其替换为 pyspark 中的异常值。

Answer 1

您可以使用when 来替换使用给定条件的异常值。要替换为均值，您可以使用mean 窗口函数而不是将其收集到变量中，并使用F.round 将其四舍五入为最接近的整数：

from pyspark.sql import functions as F, Window

upper = 75
lower = 12

df2 = df.withColumn(
    'a', 
    F.when(
        (df['a'] > upper) | (df['a'] < lower), 
        F.round(F.mean('a').over(Window.orderBy(F.lit(1)))).cast('int')
        # or you can use 
        # F.round(F.lit(df.select(F.mean(F.col('a')).alias('mean')).collect()[0]['mean'])).cast('int')
    ).otherwise(F.col('a'))
)

df2.show()
+---+---+
|  a|  b|
+---+---+
| 27|  0|
| 47|  1|
| 47|  2|
| 21|  3|
| 46|  4|
| 47|  5|
+---+---+