spark数据清洗代码通用

Apache Spark 是一个强大的大数据处理框架，用于数据清洗（Data Cleaning）的过程通常涉及到数据去重、缺失值处理、异常值检测和修正等步骤。在Spark中，你可以使用DataFrame API或RDD（Resilient Distributed Datasets）来进行数据清洗。以下是一些通用的代码示例：

1. **导入库和创建SparkSession**:

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.appName("DataCleaningExample").getOrCreate()

2. **加载数据**:

# 假设我们从HDFS或文件系统加载CSV数据
data = spark.read.format("csv").option("header", "true").load("path/to/your/data.csv")

3. **检查和处理缺失值**:

# 查看数据中是否存在缺失值
data.printSchema()  # 查看列的数据类型
data.count()       # 总行数
na_counts = data.select([col.agg({"isNotNull": "count"}).alias(col.name) for col in data.columns]).collect()  # 检查每列的非空记录数

# 处理缺失值，如用平均值填充数值列
data = data.fillna(data.mean(), subset=[column_name])  # 用列均值填充特定列

4. **数据去重**:

# 去除重复行（基于整个DataFrame）
data = data.dropDuplicates()

5. **异常值检测和处理**:

# 假设我们使用Z-score检测异常值
from pyspark.ml.feature import StandardScaler
scaler = StandardScaler(inputCol="your_column", outputCol="scaled_column")
scaled_data = scaler.fit(data).transform(data)
outliers = scaled_data.filter((F.abs(scaled_data["scaled_column"] - scaled_data["scaled_column"].mean()) > 3 * scaled_data["scaled_column"].stddev()))

# 可选择删除或替换异常值
data = data.subtract(outliers)

6. **保存清洗后的数据**:

cleaned_data.write.format("csv").save("path/to/cleaned/data.csv")