PySpark 2.2.0：“numpy.ndarray”对象没有属性“indices”

Question

任务

我正在使用 Python API for Spark (PySpark) 计算 __SparseVector__ 内索引的大小。

脚本

def score_clustering(dataframe):
assembler = VectorAssembler(inputCols = dataframe.drop("documento").columns, outputCol = "variables")
data_transformed = assembler.transform(dataframe)
data_transformed_rdd = data_transformed.select("documento", "variables").orderBy(data_transformed.documento.asc()).rdd
count_variables = data_transformed_rdd.map(lambda row : [row[0], row[1].indices.size]).toDF(["id", "frequency"])

问题

当我在 __count_variables__ 数据帧上执行操作 __.count()__ 时，会出现错误：

AttributeError: 'numpy.ndarray' 对象没有属性 'indices'

要考虑的主要部分是：

data_transformed_rdd.map(lambda row : [row[0], row[1].indices.size]).toDF(["id", "frequency"])

我相信这个块与错误有关，但如果我通过映射 __lambda expression__ 以 __SparseVector__ （创建__assembler__)。

有什么建议吗？有谁知道我做错了什么？

Answer 1

这里有两个问题。第一个在indices.size调用中，indices和size是SparseVector class的两个不同属性，size是完整的向量大小，indices是向量索引，其值为非零，但size 不是 indices 属性。所以，假设你所有的向量都是 SparseVector 类的实例：

from pyspark.ml.linalg import Vectors

df = spark.createDataFrame([(0, Vectors.sparse(4, [0, 1], [11.0, 2.0])),
                            (1, Vectors.sparse(4, [], [])),
                            (3, Vectors.sparse(4, [0,1,2], [2.0, 2.0, 2.0]))],
                           ["documento", "variables"])

df.show()

+---------+--------------------+
|documento|           variables|
+---------+--------------------+
|        0|(4,[0,1],[11.0,2.0])|
|        1|           (4,[],[])|
|        3|(4,[0,1,2],[2.0,2...|
+---------+--------------------+

解决办法是len函数：

df = df.rdd.map(lambda x: (x[0], x[1], len(x[1].indices)))\
               .toDF(["documento", "variables", "frecuencia"])
df.show()  
+---------+--------------------+----------+
|documento|           variables|frecuencia|
+---------+--------------------+----------+
|        0|(4,[0,1],[11.0,2.0])|         2|
|        1|           (4,[],[])|         0|
|        3|(4,[0,1,2],[2.0,2...|         3|
+---------+--------------------+----------+

第二个问题来了：VectorAssembler 并不总是生成 SparseVectors，取决于哪个更有效，可以生成 SparseVector 或 DenseVectors（基于原始向量具有的零的数量）。例如，假设下一个数据帧：

df = spark.createDataFrame([(0, Vectors.sparse(4, [0, 1], [11.0, 2.0])),
                             (1, Vectors.dense([1., 1., 1., 1.])),
                              (3, Vectors.sparse(4, [0,1,2], [2.0, 2.0, 2.0]))], 
                           ["documento", "variables"])

df.show()      
+---------+--------------------+
|documento|           variables|
+---------+--------------------+
|        0|(4,[0,1],[11.0,2.0])|
|        1|   [1.0,1.0,1.0,1.0]|
|        3|(4,[0,1,2],[2.0,2...|
+---------+--------------------+

文档 1 是 DenseVector 并且之前的解决方案不起作用，因为 DenseVectors 没有 indices 属性，因此您必须使用更通用的向量表示来处理包含稀疏和密集向量的 DataFrame，例如例如numpy:

import numpy as np
df = df.rdd.map(lambda x: (x[0], 
                           x[1], 
                           np.nonzero(x[1])[0].size))\
                .toDF(["documento", "variables", "frecuencia"])
df.show() 
+---------+--------------------+----------+
|documento|           variables|frecuencia|
+---------+--------------------+----------+
|        0|(4,[0,1],[11.0,2.0])|         2|
|        1|   [1.0,1.0,1.0,1.0]|         4|
|        3|(4,[0,1,2],[2.0,2...|         3|
+---------+--------------------+----------+