用平均值替换缺失值 - Spark Dataframe

Question

我有一个带有一些缺失值的 Spark Dataframe。我想通过用该列的平均值替换缺失值来执行简单的插补。我对 Spark 很陌生，所以我一直在努力实现这个逻辑。到目前为止，这是我设法做到的：

a) 要对单个列执行此操作（假设是 Col A），这行代码似乎有效：

df.withColumn("new_Col", when($"ColA".isNull, df.select(mean("ColA"))
  .first()(0).asInstanceOf[Double])
  .otherwise($"ColA"))

b) 但是，我一直无法弄清楚如何对数据框中的所有列执行此操作。我正在尝试 Map 函数，但我相信它会循环遍历数据帧的每一行

c) 关于 SO - here 也有类似的问题。虽然我喜欢这个解决方案（使用聚合表和合并），但我非常想知道是否有一种方法可以通过遍历每一列来做到这一点（我来自 R，所以使用更高阶函数遍历每一列，比如lapply 对我来说似乎更自然）。

谢谢！

Answer 1

Spark >= 2.2

您可以使用org.apache.spark.ml.feature.Imputer（它同时支持均值和中值策略）。

斯卡拉 :

import org.apache.spark.ml.feature.Imputer

val imputer = new Imputer()
  .setInputCols(df.columns)
  .setOutputCols(df.columns.map(c => s"${c}_imputed"))
  .setStrategy("mean")

imputer.fit(df).transform(df)

Python:

from pyspark.ml.feature import Imputer

imputer = Imputer(
    inputCols=df.columns, 
    outputCols=["{}_imputed".format(c) for c in df.columns]
)
imputer.fit(df).transform(df)

Spark < 2.2

给你：

import org.apache.spark.sql.functions.mean

df.na.fill(df.columns.zip(
  df.select(df.columns.map(mean(_)): _*).first.toSeq
).toMap)

哪里

df.columns.map(mean(_)): Array[Column]

计算每列的平均值，

df.select(_: *).first.toSeq: Seq[Any]

收集聚合值并将行转换为 Seq[Any]（我知道这是次优的，但这是我们必须使用的 API），

df.columns.zip(_).toMap: Map[String,Any]

创建 aMap: Map[String, Any]，它从列名映射到它的平均值，最后：

df.na.fill(_): DataFrame

使用以下方法填充缺失值：

fill: Map[String, Any] => DataFrame

来自 DataFrameNaFunctions.

要取消 NaN 个条目，您可以替换：

df.select(df.columns.map(mean(_)): _*).first.toSeq

与：

import org.apache.spark.sql.functions.{col, isnan, when}


df.select(df.columns.map(
  c => mean(when(!isnan(col(c)), col(c)))
): _*).first.toSeq

Answer 2

对于 PySpark，这是我使用的代码：

mean_dict = { col: 'mean' for col in df.columns }
col_avgs = df.agg( mean_dict ).collect()[0].asDict()
col_avgs = { k[4:-1]: v for k,v in col_avgs.iteritems() }
df.fillna( col_avgs ).show()

四个步骤是：

创建字典mean_dict 将列名映射到聚合操作（均值）
计算每一列的平均值，并将其保存为字典col_avgs
col_avgs 中的列名称以 avg( 开头并以 ) 结尾，例如avg(col1)。去掉括号。
使用 col_avgs

Answer 3

用于在 PySpark < 2.2 中估算中位数（而不是平均值）

## filter numeric cols
num_cols = [col_type[0] for col_type in filter(lambda dtype: dtype[1] in {"bigint", "double", "int"}, df.dtypes)]
### Compute a dict with <col_name, median_value>
median_dict = dict()
for c in num_cols:
   median_dict[c] = df.stat.approxQuantile(c, [0.5], 0.001)[0]

然后，应用na.fill

df_imputed = df.na.fill(median_dict)

用平均值替换缺失值 - Spark Dataframe

Replace missing values with mean - Spark Dataframe

scala

dataframe

apache-spark

apache-spark-sql

imputation