如何计算 SparseVector 中的特征存在量

Question

我有一个这样的数据框：

+---+-------+--------------+
| id|clicked|     features |
+---+-------+--------------+
|  7|  1.0  |(4,[3],[1.0]) |
|  8|  0.0  |(4,[2],[12.0])|
|  9|  0.0  |(4,[3],[0.1]) |
+---+-------+--------------+

并将其转换为 RDD：

Row(id=7, clicked=1.0, features=SparseVector(4, {3: 1.0}))
Row(id=8, clicked=0.0, features=SparseVector(4, {2: 12.0}))
Row(id=9, clicked=0.0, features=SparseVector(4, {3: 0.1}))

现在我想查找每个功能的存在数。例如，在我的数据框/rdd 中，特征列包含 5 个特征（0 到 4）。由于索引 0、1 和 4 中的特征不包含任何值，因此它们的数量为 0。并且特征索引 2 为 1，特征索引 3 为 2。

我想在字典数据结构中获取该信息。怎么做？

{0:0, 1:0, 2:1, 3:2, 4:0}

我正在使用 PySpark，但使用 Scala 的答案也可以。

Answer 1

最好的办法是在创建SparseVector 之前进行计数。如果这不可能，您基本上有两个选择（直到VectorUDTs 是easily castable into arrays）。

在这两种情况下，计算每个特征存在的值数量的方法是相同的。循环遍历SparseVector 的大小范围并检查该索引是否存在于SparseVector.indices 列表中。这将返回所有功能的计数，包括计数为 0 的功能。

一种更简单的方法是为SparseVector.indices 中的每个索引创建(index, 1) 形式的元组，但这会从最终输出中排除任何没有任何值的特征。

选项 1：定义 udf、explode 和聚合：

import pyspark.sql.functions as f

featureCount_udf = f.udf(
    lambda r: [(x, int(x in r.indices)) for x in range(r.size)],
    ArrayType(
        StructType(
            [
                StructField("featureNumber", IntegerType()),
                StructField("count", IntegerType())
            ]
        )
    )
)

df.select(f.explode(featureCount_udf("features")).alias("features"))\
    .select("features.*")\
    .groupBy("featureNumber")\
    .agg(f.sum("count").alias("count"))\
    .show()
#+-------------+-----+
#|featureNumber|count|
#+-------------+-----+
#|            0|    0|
#|            2|    1|
#|            1|    0|
#|            3|    2|
#+-------------+-----+

选项2：转换为rdd和flatMap：

from operator import add

df.select("features")\
    .rdd\
    .flatMap(
        lambda r: [(x, int(x in r["features"].indices)) for x in range(r["features"].size)]
    )\
    .reduceByKey(add)\
    .toDF(["featureNumber", "count"])\
    .show()
#+-------------+-----+
#|featureNumber|count|
#+-------------+-----+
#|            0|    0|
#|            2|    1|
#|            1|    0|
#|            3|    2|
#+-------------+-----+

在这里，我们将 flatMap 每个 row 转换为 (featureNumber, containsValue) 形式的元组。然后我们可以调用reduceByKey为每个特征添加指标变量。

---

原答案

如果你想在字典中输出，你将不得不在某个时候调用collect()。

data = df.select("features").collect()

现在您拥有pyspark.sql.Rows 列表中的数据，您可以遍历并使用.indices 和.size 来识别哪些列具有值。

print([[int(x in r["features"].indices) for x in range(r["features"].size)] for r in data])
#[[0, 0, 0, 1], [0, 0, 1, 0], [0, 0, 0, 1]]

由此您可以创建一个numpy 数组并对列求和。最后在结果上调用 enumerate 并将其传递给 dict 构造函数。

把它们放在一起：

mydict = dict(
    enumerate(
        np.array(
            [[int(x in r["features"].indices) for x in range(r["features"].size)]
             for r in data]
        ).sum(0)
    )
)
print(mydict)
#{0: 0, 1: 0, 2: 1, 3: 2}