machine-learning - PySpark 中的 KMeans 聚类

Question

我有一个包含许多列的 Spark 数据框“mydataframe”。我试图仅在两列上运行 kmeans:纬度和经度(纬度和经度)，使用它们作为简单值)。我想仅基于这两列提取 7 个簇，然后我想将簇分配附加到我的原始数据帧。我试过:

from numpy import array
from math import sqrt
from pyspark.mllib.clustering import KMeans, KMeansModel

# Prepare a data frame with just 2 columns:
data = mydataframe.select('lat', 'long')
data_rdd = data.rdd  # needs to be an RDD
data_rdd.cache()

# Build the model (cluster the data)
clusters = KMeans.train(data_rdd, 7, maxIterations=15, initializationMode="random")

但过了一会儿我收到错误:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 5191.0 failed 4 times, most recent failure: Lost task 1.3 in stage 5191.0 (TID 260738, 10.19.211.69, executor 1): org.apache.spark.api.python.PythonException: Traceback (most recent call last)

我尝试分离并重新附加集群。相同的结果。我做错了什么？

Answer 1

因为，基于 another recent question of yours ，我猜您正处于 Spark 集群的第一步(您甚至导入 sqrt 和 array，可能从未使用过它们因为它就像 docs example 中的那样)，让我在更一般的层面上提供建议，而不是在您在这里提出的具体问题中(希望也能让您免于随后提出 3-4 个问题，试图恢复集群分配)进入您的数据框)...

自

1. 您的数据已在数据框中

2. 您想要将集群成员资格附加回您的初始数据框

您没有理由恢复到 RDD 并使用 (soon to be deprecated) MLlib 包；使用(现在推荐的)ML 包，您可以更轻松、更优雅、更高效地完成工作，该包直接与数据帧配合使用。

第 0 步 - 制作一些与您相似的玩具数据:

spark.version
# u'2.2.0'

df = spark.createDataFrame([[0, 33.3, -17.5],
                              [1, 40.4, -20.5],
                              [2, 28., -23.9],
                              [3, 29.5, -19.0],
                              [4, 32.8, -18.84]
                             ],
                              ["other","lat", "long"])

df.show()
# +-----+----+------+
# |other| lat|  long|
# +-----+----+------+
# |    0|33.3| -17.5|
# |    1|40.4| -20.5| 
# |    2|28.0| -23.9|
# |    3|29.5| -19.0|
# |    4|32.8|-18.84|
# +-----+----+------+

第 1 步 - 组装您的功能

与大多数机器学习包相比，Spark ML 要求将输入特征收集在数据帧的单列中，通常命名为features；它提供了执行此操作的具体方法，VectorAssembler:

from pyspark.ml.feature import VectorAssembler

vecAssembler = VectorAssembler(inputCols=["lat", "long"], outputCol="features")
new_df = vecAssembler.transform(df)
new_df.show()
# +-----+----+------+-------------+ 
# |other| lat|  long|     features|
# +-----+----+------+-------------+
# |    0|33.3| -17.5| [33.3,-17.5]|
# |    1|40.4| -20.5| [40.4,-20.5]|
# |    2|28.0| -23.9| [28.0,-23.9]| 
# |    3|29.5| -19.0| [29.5,-19.0]|
# |    4|32.8|-18.84|[32.8,-18.84]|
# +-----+----+------+-------------+ 

正如可能已经猜到的，参数 inputCols 用于告诉 VectoeAssembler 我们的数据框中的哪些特定列将用作特征。

第 2 步 - 拟合您的 KMeans 模型

from pyspark.ml.clustering import KMeans

kmeans = KMeans(k=2, seed=1)  # 2 clusters here
model = kmeans.fit(new_df.select('features'))

select('features') 此处用于告诉算法使用数据帧的哪一列进行聚类 - 请记住，在上述步骤 1 之后，您的原始 lat & long 特征不再直接使用。

第 3 步 - 转换初始数据框以包含聚类分配

transformed = model.transform(new_df)
transformed.show()    
# +-----+----+------+-------------+----------+ 
# |other| lat|  long|     features|prediction|
# +-----+----+------+-------------+----------+
# |    0|33.3| -17.5| [33.3,-17.5]|         0| 
# |    1|40.4| -20.5| [40.4,-20.5]|         1|
# |    2|28.0| -23.9| [28.0,-23.9]|         0|
# |    3|29.5| -19.0| [29.5,-19.0]|         0|
# |    4|32.8|-18.84|[32.8,-18.84]|         0|
# +-----+----+------+-------------+----------+

转换数据帧的最后一列，预测，显示了集群分配 - 在我的玩具案例中，我最终在集群 #0 和 1 中得到了 4 条记录记录在集群 #1 中。

您可以使用select语句进一步操作转换后的数据框，甚至删除features列(现在已完成其功能，可能不再需要)...

希望您现在更接近您最初真正想要实现的目标。对于提取集群统计信息等，another recent answer of mine 可能会有所帮助......