我有一个名为’mydataframe’的Spark数据框，包含许多列。我试图对其中的两个列：lat和long（纬度和经度）进行KMeans聚类，将它们作为简单值。我希望基于这两个列提取7个聚类，然后将聚类分配结果附加到我的原始数据框中。我尝试了以下代码：
from numpy import arrayfrom math import sqrtfrom pyspark.mllib.clustering import KMeans, KMeansModel# 准备只包含2列的数据框:data = mydataframe.select('lat', 'long')data_rdd = data.rdd # 需要转换为RDDdata_rdd.cache()# 构建模型（对数据进行聚类）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)
我尝试了分离和重新附加聚类。结果还是一样。我做错了什么？
回答：
基于你最近的另一个问题，我猜你正处于Spark聚类的初级阶段（你甚至导入了sqrt和array，但从未使用它们，可能是因为文档示例中是这样做的），让我提供一些更普遍的建议，而不是你在这里提出的具体问题（希望这也能避免你随后再提出3-4个问题，试图将聚类分配结果重新整合到你的数据框中）…
由于
-
你的数据已经在一个数据框中
-
你想将聚类成员信息附加回你的初始数据框
你没有理由回退到RDD并使用（即将被废弃的）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 VectorAssemblervecAssembler = 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 KMeanskmeans = KMeans(k=2, seed=1) # 这里是2个聚类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显示了聚类分配结果 – 在我的玩具案例中，我最终得到了4个记录在聚类#0，1个记录在聚类#1。
你可以进一步使用select语句来操作transformed数据框，甚至可以drop掉features列（它现在已经完成了它的功能，可能不再需要）…
希望你现在离你最初想要实现的目标更近了一步。对于提取聚类统计数据等，我的另一个最近的回答可能会有帮助…
