PySpark 中的 KMeans 聚类

Question

我有一个包含许多列的 spark 数据框“mydataframe”。我试图只在两列上运行 kmeans：lat 和 long（纬度和经度），使用它们作为简单值）。我想仅基于这 2 列提取 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：作业因阶段故障而中止：阶段 5191.0 中的任务 1 失败 4 次，最近一次失败：阶段 5191.0 中丢失任务 1.3（TID 260738、10.19.211.69、执行程序 1）：org .apache.spark.api.python.PythonException: Traceback（最近一次调用最后一次）

我已尝试分离并重新连接集群。结果相同。我做错了什么？

Answer 1

因为，基于another recent question of yours，我猜您是 Spark 集群的第一步（您甚至导入了 sqrt 和 array，但从未使用过它们，可能是因为它就像在 @ 987654322@），让我在更一般的层面上提供建议，而不是在您在这里提出的具体问题中提供建议（希望也可以让您免于随后打开 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 步 - 组装您的功能

与现有的大多数 ML 包相比，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|
# +-----+----+------+-------------+----------+

transformed 数据框的最后一列 prediction 显示了集群分配 - 在我的玩具案例中，我在集群 #0 中有 4 条记录，在集群 #1 中有 1 条记录。

您可以使用select 语句进一步操作transformed 数据框，甚至可以使用drop 的features 列（现在已完成其功能，可能不再需要）...

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

Answer 2

尽管我给出了其他一般性的回答，并且如果您出于某种原因必须坚持使用 MLlib 和 RDD，这就是导致您使用同一个玩具 df 时出错的原因。

当您将数据框中的select 列转换为 RDD 时，正如您所做的那样，结果是 Rows 的 RDD：

df.select('lat', 'long').rdd.collect()
# [Row(lat=33.3, long=-17.5), Row(lat=40.4, long=-20.5), Row(lat=28.0, long=-23.9), Row(lat=29.5, long=-19.0), Row(lat=32.8, long=-18.84)]

不适合作为 MLlib KMeans 的输入。您需要map 操作才能使其工作：

df.select('lat', 'long').rdd.map(lambda x: (x[0], x[1])).collect()
# [(33.3, -17.5), (40.4, -20.5), (28.0, -23.9), (29.5, -19.0), (32.8, -18.84)]

所以，你的代码应该是这样的：

from pyspark.mllib.clustering import KMeans, KMeansModel

rdd = df.select('lat', 'long').rdd.map(lambda x: (x[0], x[1]))
clusters = KMeans.train(rdd, 2, maxIterations=10, initializationMode="random") # works OK
clusters.centers
# [array([ 40.4, -20.5]), array([ 30.9 , -19.81])]