我有一个包含一些缺失值的 Spark 数据框。我想通过用该列的平均值替换缺失值来执行简单的插补。我对 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 对我来说似乎更自然)。
谢谢!
最佳答案
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
关于scala - 用平均值替换缺失值 - Spark Dataframe,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40057563/