PySpark中的所有与所有比较

Question

我对PySpark相对来说比较新。

我有2M火车实例和4M测试实例。每个实例都由一个id和一个特征向量表示。我有一个自定义函数来计算两对实例之间的相似度。

作为一个输出，我希望每个测试实例都有一个排序的列车实例。这是一个很常见的问题，但实例的数量是挑战。

由于，我必须做的所有-VS-所有比较，我在PySpark使用的cartesian产品功能如下：

train_testpairs = trainRDD.cartesian(testRDD) 
train_testpairs_simscores = train_testpairs.map(myscore) 

在，train_testpairs_simscores，我有以下类型的元组：

(train_id,test_id,score) 

但是，现在我想火车实例为每个测试之类的实例的排序顺序：

test_id1,[train_id432, train_id832,.....] 

我明白，对于每个test_id，这将有一个2M大小的列表。因此，这个列表的n％也是可以的。可能是最接近测试ID的前1000个列车实例。

我该如何做到这一点？

我试图按testID分组，然后在每个列表中排序，但groupBy本身太耗时。 任何帮助，将不胜感激。

Answer 1

面临的一个非常类似的问题，我结束了“手动”计算相似性得分，而不是使用cartesian()：

def f(train_instance): 
    return train_instance,sorted(myscore(train_instance,test_instance) for test_instance in TEST_INSTANCES) 

trainRDD.map(f) 

这样你避免扩大2米*4米项目的笛卡尔积，然后重新组合它们