由于 Spark 的 mllib 没有最近邻居功能,我正在尝试使用Annoy https://github.com/spotify/annoy为近似最近邻。我尝试广播 Annoy 对象并将其传递给工人;然而,它并没有按预期运行。
下面是可重复性的代码(在 PySpark 中运行)。使用带有 Spark 的 Annoy 与不带有 Spark 的 Annoy 时的差异突出显示了该问题。
from annoy import AnnoyIndex
import random
random.seed(42)
f = 40
t = AnnoyIndex(f) # Length of item vector that will be indexed
allvectors = []
for i in xrange(20):
v = [random.gauss(0, 1) for z in xrange(f)]
t.add_item(i, v)
allvectors.append((i, v))
t.build(10) # 10 trees
# Use Annoy with Spark
sparkvectors = sc.parallelize(allvectors)
bct = sc.broadcast(t)
x = sparkvectors.map(lambda x: bct.value.get_nns_by_vector(vector=x[1], n=5))
print "Five closest neighbors for first vector with Spark:",
print x.first()
# Use Annoy without Spark
print "Five closest neighbors for first vector without Spark:",
print(t.get_nns_by_vector(vector=allvectors[0][1], n=5))
看到的输出:
Spark 第一个向量的五个最近邻居:无
没有 Spark 的第一个向量的五个最近邻:[0, 13, 12, 6, 4]
我从未使用过 Annoy,但我很确定包描述解释了这里发生的情况:
它还创建基于文件的大型只读数据结构,这些数据结构被映射到内存中,以便许多进程可以共享相同的数据。
由于当您序列化它并将其传递给工作人员时它使用内存映射索引,因此所有数据都会丢失。
尝试这样的事情:
from pyspark import SparkFiles
t.save("index.ann")
sc.addPyFile("index.ann")
def find_neighbors(iter):
t = AnnoyIndex(f)
t.load(SparkFiles.get("index.ann"))
return (t.get_nns_by_vector(vector=x[1], n=5) for x in iter)
sparkvectors.mapPartitions(find_neighbors).first()
## [0, 13, 12, 6, 4]
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)