17

我有一个带有一些缺失值的 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 对我来说似乎更自然）。

谢谢！

4

3 回答 3

27

火花 >= 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)

蟒蛇：

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)

火花 < 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
于 2016-10-15T13:14:43.163 回答
4

对于 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()

这四个步骤是：

  1. 创建字典mean_dict映射列名到聚合操作（平均值）
  2. 计算每列的平均值，并将其保存为字典col_avgs
  3. col_avgs以开头avg(和结尾的列名)，例如avg(col1)。去掉括号。
  4. 使用平均值填充数据框的列col_avgs
于 2017-03-23T20:54:40.183 回答
4

用于估算 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)
于 2017-09-09T19:29:24.543 回答