Pyspark：K 表示有距离或偏差的结果？

Question

来自https://spark.apache.org/docs/2.2.0/ml-clustering.html#k-means

我知道在 之后kmModel.transform(df)，prediction输出数据框有一列说明记录/点属于哪一列。

但是，我也想知道每个记录/点如何偏离质心，所以我知道这个集群中的哪些点是典型的，以及集群之间可能存在什么

我该怎么做？默认情况下似乎没有由包实现

谢谢！

Answer 1

假设我们有以下样本数据和 kmeans 模型：

from pyspark.ml.linalg import Vectors
from pyspark.ml.clustering import KMeans
import pyspark.sql.functions as F

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]),),
        (Vectors.dense([10.0, 1.5]),), (Vectors.dense([11, 0.0]),) ]
df = spark.createDataFrame(data, ["features"])

n_centres = 2
kmeans = KMeans().setK(n_centres).setSeed(1)
kmModel = kmeans.fit(df)
df_pred = kmModel.transform(df)
df_pred.show()

+----------+----------+
|  features|prediction|
+----------+----------+
| [0.0,0.0]|         1|
| [1.0,1.0]|         1|
| [9.0,8.0]|         0|
| [8.0,9.0]|         0|
|[10.0,1.5]|         0|
|[11.0,0.0]|         0|
+----------+----------+

现在，让我们添加一个包含中心坐标的列：

l_clusters = kmModel.clusterCenters()
# Let's convert the list of centers to a dict, each center is a list of float
d_clusters = {int(i):[float(l_clusters[i][j]) for j in range(len(l_clusters[i]))] 
              for i in range(len(l_clusters))}

# Let's create a dataframe containing the centers and their coordinates
df_centers = spark.sparkContext.parallelize([(k,)+(v,) for k,v in 
d_clusters.items()]).toDF(['prediction','center'])

df_pred = df_pred.withColumn('prediction',F.col('prediction').cast(IntegerType()))
df_pred = df_pred.join(df_centers,on='prediction',how='left')
df_pred.show()


+----------+----------+------------+
|prediction|  features|      center|
+----------+----------+------------+
|         0| [8.0,9.0]|[9.5, 4.625]|
|         0|[10.0,1.5]|[9.5, 4.625]|
|         0| [9.0,8.0]|[9.5, 4.625]|
|         0|[11.0,0.0]|[9.5, 4.625]|
|         1| [1.0,1.0]|  [0.5, 0.5]|
|         1| [0.0,0.0]|  [0.5, 0.5]|
+----------+----------+------------+

最后，我们可以使用 udf 来计算列特征和中心坐标之间的距离：

get_dist = F.udf(lambda features, center : 
                 float(features.squared_distance(center)),FloatType())
df_pred = df_pred.withColumn('dist',get_dist(F.col('features'),F.col('center')))
df_pred.show()

+----------+----------+------------+---------+
|prediction|  features|      center|     dist|
+----------+----------+------------+---------+
|         0|[11.0,0.0]|[9.5, 4.625]|23.640625|
|         0| [9.0,8.0]|[9.5, 4.625]|11.640625|
|         0| [8.0,9.0]|[9.5, 4.625]|21.390625|
|         0|[10.0,1.5]|[9.5, 4.625]|10.015625|
|         1| [1.0,1.0]|  [0.5, 0.5]|      0.5|
|         1| [0.0,0.0]|  [0.5, 0.5]|      0.5|
+----------+----------+------------+---------+