pyspark kmeansアルゴリズムの結果をデータフレームに追加しますか？

Question

私は地理情報を含むスパークデータフレームを持っています。

my_df.show(2) 

## +----+----+-----------+----------+ 
## | x0 | x1 | longitude | latitude | 
## +----+----+-----------+----------+ 
## | ...| ...| 51.043 | 13.6847 | 
## | ...| ...| 42.6753 | 23.3218 | 

私は私のデータフレームのうち、経度と緯度を取り、pysparkから関数kmeansライブラリといくつかの中心点をcaluculated。

#Trains a k-means model 
k = 120 
model = KMeans.train(dataset, k) 
print ("Final centers: " + str(model.clusterCenters)) 

出力

Final centers: [array([ 51.04307692, 13.68474126]), array([-33.434  , -70.58366667]), array([ 42.67533333, 23.32185981]), array([ 45.876, -61.492]), array([ 53.07465714, 8.4655 ]), array([ 4.594, 114.262]), array([ 48.15665306, 11.54269728]), array([ 51.51729851, 7.49838806]), array([ 48.76316125, 9.15357859]), .... 

誰でも自分のデータフレームにマッチングセンターを追加する方法を考えて？

## +----+----+-----------+----------+-----------+----------+ 
## | x0 | x1 | longitude | latitude | mean_long | mean_lat | 
## +----+----+-----------+----------+-----------+----------+ 
## | ...| ...| 51.043 | 13.6847 | 50.000 | 15.000 | 
## | ...| ...| 42.6753 | 23.3218 | 50.000 | 15.000 | 

Answer 1

あなたはレガシーpyspark.mllib、新しいpyspark.ml APIを使用する必要がありませんDataFrames使用することを決めた場合。 K-Meansを含むいくつかのクラスタリング方法を提供し、その予測方法は予測列をDataFrameに添付します。

詳細については、（APIをし、入力タイプが必要）MLのマニュアルを参照してください：

• https://spark.apache.org/docs/latest/ml-clustering.html#k-means

Answer 2

・ホープ、このことができます！
（ノート - 私はスパークドキュメントページからサンプルデータをとっている）

from pyspark.ml.linalg import Vectors 
from pyspark.ml.clustering import KMeans 
import pandas as pd 

#generate data 
data = [(Vectors.dense([0.0, 0.0]),), (Vectors.dense([1.0, 1.0]),), 
     (Vectors.dense([9.0, 8.0]),), (Vectors.dense([8.0, 9.0]),)] 
df = sqlContext.createDataFrame(data, ["features"]) 
df.show() 

#run kmeans clustering model 
kmeans = KMeans(k=2, seed=1) 
model = kmeans.fit(df) 
predictions=model.transform(df).withColumnRenamed("prediction","cluster_id") 

centers = model.clusterCenters() 
#preprocessing centers so that it can be joined with predictions dataframe 
centers_p_df = pd.DataFrame(centers) 
centers_p_df.insert(0, 'new_col', range(0, len(centers_p_df))) 
centers_df = sqlContext.createDataFrame(centers_p_df, schema=['cluster_id','centers_col1','centers_col2']) 

final_df = predictions.join(centers_df, on="cluster_id").drop("cluster_id") 
final_df.show()