如何在Spark中以DenseVector为键按键分组RDD？

Question

{{cdd}和一对被创建的值都是一个值。e、 g

[(DenseVector([3,4]),10),  (DenseVector([3,4]),20)]

现在我想按键k1:DenseVector([3,4])分组。我希望该行为将键k1的所有值分组，它们是10和{}。但我得到的结果是

^{pr2}$

而不是

[(DenseVector([3,4]), [10,20])]

如果我遗漏了什么，请告诉我。在

代码是：

#simplified version of code
#rdd1 is an rdd containing [(DenseVector([3,4]),10),  (DenseVector([3,4]),20)]
rdd1.groupByKey().map(lambda x : (x[0], list(x[1])))
print(rdd1.collect())

Answer 1

好吧，这是一个棘手的问题，简短的回答是你不能。为了理解为什么你必须深入研究DenseVector实现。DenseVector只是NumPy float64ndarray的包装

>>> dv1 = DenseVector([3.0, 4.0])
>>> type(dv1.array)
<type 'numpy.ndarray'>
>>> dv1.array.dtype
dtype('float64')

由于NumPy ndarrays，不像{}是可变的，所以不能以有意义的方式散列，尽管有趣的是提供了__hash__方法。有一个有趣的问题涵盖了这个问题（参见：numpy ndarray hashability）。在

^{pr2}$

DenseVector从object继承{}方法，它只是基于id（给定实例的内存地址）：

>>> id(d1) / 16 == hash(d1)
True

不幸的是，这意味着具有相同内容的两个DenseVectors具有不同的哈希：

>>> dv2 = DenseVector([3.0, 4.0])
>>> hash(dv1) == hash(dv2)
False

你能做什么？最简单的方法是使用不可变的数据结构，它提供一致的hash实现，例如元组：

rdd.groupBy(lambda (k, v): tuple(k))

注意：在实践中，使用数组作为键很可能是个坏主意。对于大量的元素，散列过程可能是非常昂贵的有用的。不过，如果您真的需要这样的东西，Scala似乎可以正常工作：

import org.apache.spark.mllib.linalg.Vectors

val rdd = sc.parallelize(
    (Vectors.dense(3, 4), 10) :: (Vectors.dense(3, 4), 20) :: Nil)
rdd.groupByKey.collect