我有一个包含许多列的spark数据帧'mydataframe'。我试图只在两列上运行kmeans：lat和long（纬度和经度），使用它们作为简单值）。我想基于这两个列提取7个集群，然后我想将集群asignment附加到我的原始数据帧。我试过了：
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（最近一次调用最后一次）
我试图分离并重新连接群集。结果相同。我究竟做错了什么？
非常感谢你！
因为，基于another recent question of yours，我猜你正处于Spark群集的第一步（你甚至导入sqrt和array，没有使用它们，可能是因为它就像在docs example中那样），让我提供更多建议一般水平，而不是你在这里问的具体问题（希望还能让你免于打开3-4个问题，尝试将你的群集分配回到你的数据框中）......
以来
你没有理由恢复到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条记录。
您可以使用transformed语句进一步操纵select数据框，甚至drop features列（现已完成其功能，可能不再需要）...
希望你现在更接近你最初想要实现的目标。为了提取集群统计数据等，another recent answer of mine可能会有所帮助......
尽管我的其他一般性答案，如果你，无论出于何种原因，必须坚持MLlib和RDDs，这是导致你的错误使用相同的玩具df。
当你从数据帧中select列转换为RDD时，结果是行的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])]