如何将类型Row转换为Vector以提供给KMeans

Question

当我尝试将df2提供给kmeans时,我收到以下错误

clusters = KMeans.train(df2, 10, maxIterations=30,
                        runs=10, initializationMode="random")

我得到的错误:

Cannot convert type <class 'pyspark.sql.types.Row'> into Vector

df2是一个如下创建的数据框:

df = sqlContext.read.json("data/ALS3.json")
df2 = df.select('latitude','longitude')

df2.show()


     latitude|       longitude|

   60.1643075|      24.9460844|
   60.4686748|      22.2774728|

如何将这两列转换为Vector并将其提供给KMeans？

Answer 1

ML

问题是您错过了文档的示例,并且很明显该方法train需要DataFrame具有Vectoras功能.

要修改当前数据的结构,可以使用VectorAssembler.在你的情况下,它可能是这样的:

from pyspark.sql.functions import *

vectorAssembler = VectorAssembler(inputCols=["latitude", "longitude"],
                                  outputCol="features")

# For your special case that has string instead of doubles you should cast them first.
expr = [col(c).cast("Double").alias(c) 
        for c in vectorAssembler.getInputCols()]

df2 = df2.select(*expr)
df = vectorAssembler.transform(df2)

此外,您还应该features使用MinMaxScaler类来规范化以获得更好的结果.

MLLib

为了实现这一目的,MLLib首先需要使用map函数,将所有string值转换Double为DenseVector并将它们合并在一起.

rdd = df2.map(lambda data: Vectors.dense([float(c) for c in data]))

在此之后,您可以使用变量训练MLlib的KMeans模型rdd.

Answer 2

我让PySpark 2.3.1在DataFrame上执行KMeans，如下所示：

1. 写下要包含在聚类分析中的列的列表：

feat_cols = ['latitude','longitude']`

2. 您需要所有列都是数值：

expr = [col(c).cast("Double").alias(c) for c in feat_cols]
df2 = df2.select(*expr)

3. 创建您的特征向量mllib.linalg.Vectors：

from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=feat_cols, outputCol="features")
df3 = assembler.transform(df2).select('features')

4. 您应该规范化您的功能，因为规范化并不总是必需的，但它很少有伤害（更多关于这里）：

from pyspark.ml.feature import StandardScaler
scaler = StandardScaler(
    inputCol="features",
    outputCol="scaledFeatures",
    withStd=True,
    withMean=False)
scalerModel = scaler.fit(df3)
df4 = scalerModel.transform(df3).drop('features')\
                     .withColumnRenamed('scaledFeatures', 'features')

5. 将您的DataFrame对象df4转换为密集向量RDD：

from pyspark.mllib.linalg import Vectors
data5 = df4.rdd.map(lambda row: Vectors.dense([x for x in row['features']]))

6. 使用获得的 RDD 对象作为 KMeans 训练的输入：

from pyspark.mllib.clustering import KMeans
model = KMeans.train(data5, k=3, maxIterations=10)

7. 示例：对向量空间中的点p进行分类：

prediction = model.predict(p)