python - 如何在 PySpark 中比较两个 LabeledPoint？

Question

我有两个 LabeledPoints - Prediction1 和 Prediction2。这两个LabeledPoints都具有作为第一个元素的值和作为第二个元素的预测。我想检查 Prediction1 中的 first 元素是否等于 Prediction2 中的 first 元素。所以像这样:

for each value in Prediction1 and Prediction2:
     if Prediction1.tup[0] != Prediction2.tup[0]:
         print 'Value unequal'
         break

示例:

假设以下是LabeledPoints Prediction1的RDD:

[(1,2),(3,4),(5,6)]

预测2:

[(1,12),(3,13),(5,2)]

在上面的示例中，Prediction1(1,3,5) 的每个 LabeledPoint 的第一个元素等于预测2(1,3,5)。但如果其中之一不匹配，那么我想退出该过程并打印它们不匹配并结束。

如何在 PySpark 中做到这一点

Answer 1

假设两个 RDD 具有相同数量的分区和每个分区的元素数，您可以简单地 zip 和 take:

prediction1 = sc.parallelize([(1, 2), (3, 4), (5, 6)])
prediction2 = sc.parallelize([(1, 12), (3, 13), (5, 2)])
prediction3 = sc.parallelize([(1, 0), (5, 0), (5, 0)])

def mismatch(rdd1, rdd2):
    def mismatch_(xy):
        (x1, _), (y1, _) = xy
        return x1 != y1
    return bool(rdd1.zip(rdd2).filter(mismatch_).take(1))

mismatch(prediction1, prediction2)       
## False
mismatch(prediction1, prediction3)
## True

由于 take 是惰性的，因此它应该或多或少地按照您的预期工作。请参阅Lazy foreach on a Spark RDD

如果不满足初始条件，您可以通过组合 zipWithIndex、swap (lambda kv: (kv[1], kv[0 ])) 并加入。