如何使用PySpark对多组数据执行K-Means聚类分析

在大数据场景下，单节点机器学习工具难以处理海量多组数据的聚类需求，PySpark基于Spark MLlib提供的K-Means实现，支持分布式计算，能够高效完成多组数据的聚类分析任务。本文将完整演示从多组数据加载到聚类结果输出的全流程实现。

环境准备与依赖导入

首先需要确保本地或集群环境已经安装PySpark，本文使用的PySpark版本为3.4.0，相关依赖导入代码如下：

from pyspark.sql import SparkSession
from pyspark.ml.clustering import KMeans
from pyspark.ml.feature import VectorAssembler
from pyspark.sql.functions import col
from pyspark.ml.evaluation import ClusteringEvaluator

多组数据加载与合并

假设我们有三组不同来源的用户行为数据，分别存储为CSV文件，每组数据包含用户ID、访问时长、点击次数、购买金额四个字段，多组数据可能存在字段顺序不一致的情况，需要先统一字段后合并。

创建Spark会话

首先初始化Spark会话，这是所有PySpark操作的入口：

spark = SparkSession.builder 
    .appName("MultiGroupKMeans") 
    .master("local[*]") 
    .getOrCreate()

加载多组数据

依次加载三组CSV数据，统一字段名称后合并为单个DataFrame：

# 加载第一组用户行为数据
group1_df = spark.read.csv("data/group1_user_behavior.csv", header=True, inferSchema=True) 
    .withColumnRenamed("user_id", "userId") 
    .withColumnRenamed("visit_time", "visitTime") 
    .withColumnRenamed("click_count", "clickCount") 
    .withColumnRenamed("buy_amount", "buyAmount")

# 加载第二组用户行为数据
group2_df = spark.read.csv("data/group2_user_behavior.csv", header=True, inferSchema=True) 
    .withColumnRenamed("uid", "userId") 
    .withColumnRenamed("time", "visitTime") 
    .withColumnRenamed("clicks", "clickCount") 
    .withColumnRenamed("amount", "buyAmount")

# 加载第三组用户行为数据
group3_df = spark.read.csv("data/group3_user_behavior.csv", header=True, inferSchema=True) 
    .withColumnRenamed("id", "userId") 
    .withColumnRenamed("duration", "visitTime") 
    .withColumnRenamed("click_num", "clickCount") 
    .withColumnRenamed("pay", "buyAmount")

# 合并多组数据
all_data_df = group1_df.union(group2_df).union(group3_df)
# 查看合并后数据的基本信息
all_data_df.printSchema()
all_data_df.show(5)

数据预处理

K-Means算法要求输入为数值型的特征向量，因此需要将多个特征列合并为单个向量列，同时处理可能存在的缺失值。

特征向量化

使用VectorAssembler将访问时长、点击次数、购买金额三个特征合并为向量列：

# 定义特征列
feature_cols = ["visitTime", "clickCount", "buyAmount"]
# 创建向量组装器
assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
# 转换数据，生成特征向量列
vector_df = assembler.transform(all_data_df)

# 填充缺失值，使用各列的均值填充
from pyspark.sql.functions import avg
for col_name in feature_cols:
    mean_val = vector_df.select(avg(col(col_name))).collect()[0][0]
    vector_df = vector_df.fillna(mean_val, subset=[col_name])

# 查看向量化后的数据
vector_df.select("userId", "features").show(5, truncate=False)

K-Means模型训练

初始化K-Means模型，设置聚类数量、迭代次数等参数，使用预处理后的数据训练模型。

# 初始化K-Means模型，设置聚类数为3，最大迭代次数为20
kmeans = KMeans(k=3, maxIter=20, seed=42, featuresCol="features", predictionCol="prediction")
# 训练模型
model = kmeans.fit(vector_df)

聚类结果评估与输出

使用聚类评估器计算轮廓系数，评估聚类效果，同时输出每个样本所属的聚类类别。

聚类效果评估

# 使用训练好的模型对数据进行预测
predictions = model.transform(vector_df)
# 初始化聚类评估器，计算轮廓系数
evaluator = ClusteringEvaluator(featuresCol="features", predictionCol="prediction", metricName="silhouette")
silhouette_score = evaluator.evaluate(predictions)
print(f"聚类轮廓系数为: {silhouette_score}")

结果输出与查看

输出每个用户所属的聚类类别，同时查看每个聚类的中心特征：

# 查看每个用户的聚类结果
predictions.select("userId", "visitTime", "clickCount", "buyAmount", "prediction").show(10)

# 查看聚类中心
centers = model.clusterCenters()
for i, center in enumerate(centers):
    print(f"第{i}个聚类中心的特征值为: 访问时长={center[0]:.2f}, 点击次数={center[1]:.2f}, 购买金额={center[2]:.2f}")

注意事项

• 多组数据合并时需要确保特征字段的含义和量纲一致，若存在量纲差异较大，建议先对特征进行标准化处理。
• K值的选择可以通过肘部法则或者轮廓系数法确定，本文示例直接指定为3，实际场景中需要根据业务需求调整。
• 如果数据量较大，可以将master参数设置为集群地址，利用分布式资源提升计算效率。

最后使用完成后需要关闭Spark会话释放资源：

spark.stop()

PySparkK-Means聚类多组数据处理Spark_MLlib分布式计算修改时间：2026-07-10 20:24:28