PySpark 在Spark/Python中填充缺失值

在本文中，我们将介绍如何使用PySpark填充Spark/Python中的缺失值。缺失值是数据处理中常见的问题之一，当数据集中存在缺失值时，会影响分析和建模的准确性和可靠性。因此，填充缺失值是数据预处理的重要步骤之一。

什么是缺失值

缺失值是指数据集中缺少某些值或包含无效的值。在Spark/Python中，缺失值通常表示为null或NaN（非数字）。

常用的缺失值填充方法

在填充缺失值之前，我们需要先了解一些常用的缺失值填充方法，以便选择最适合我们数据集的方法。

1. 删除缺失值行

最简单的方法是直接删除包含缺失值的行。这种方法适用于缺失值较少的情况，但会导致数据量的减少。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MissingValues").getOrCreate()

# 仅保留不包含缺失值的行
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = df.dropna()
df.show()

2. 使用统计量填充缺失值

另一种常用的方法是使用统计量（例如平均值、中位数或众数）来填充缺失值。这种方法适用于数值型数据。

from pyspark.sql import SparkSession
from pyspark.ml.feature import Imputer

spark = SparkSession.builder.appName("MissingValues").getOrCreate()

# 使用Imputer将缺失值替换为平均值
imputer = Imputer(
    inputCols=["age", "income"],
    outputCols=["age_imputed", "income_imputed"]
).setStrategy("mean")

df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = imputer.fit(df).transform(df)
df.show()

3. 使用插值法填充缺失值

插值法是一种基于已有值推断缺失值的方法。常见的插值方法包括线性插值、多项式插值和样条插值。这种方法适用于数值型数据。

from pyspark.sql import SparkSession
from pyspark.ml.feature import Imputer

spark = SparkSession.builder.appName("MissingValues").getOrCreate()

# 使用Imputer将缺失值进行插值
imputer = Imputer(
    inputCols=["age", "income"],
    outputCols=["age_imputed", "income_imputed"]
).setStrategy("interpolate")

df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = imputer.fit(df).transform(df)
df.show()

使用PySpark填充缺失值

在PySpark中，我们可以使用DataFrame API提供的函数来填充缺失值。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MissingValues").getOrCreate()

# 填充空白值为0
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = df.fillna(0)
df.show()

如果我们需要根据列的平均值来填充缺失值，可以使用DataFrame API的agg()和fill()函数。

from pyspark.sql import SparkSession
from pyspark.sql.functions import avg

spark = SparkSession.builder.appName("MissingValues").getOrCreate()

# 根据列的平均值填充缺失值
df = spark.read.csv("data.csv", header=True, inferSchema=True)

# 计算列的平均值
means = df.agg(avg("age"), avg("income")).first()

# 使用列的平均值填充缺失值
df = df.fillna(means["avg(age)"], subset=["age"])
df = df.fillna(means["avg(income)"], subset=["income"])
df.show()