scala - Spark RDD any() 和 all() 方法？

Question

我有一个RDD[T]和一个谓词T => Boolean。我如何计算所有项目是否适合/不适合谓词？

当然我可以这样做：

rdd
 .map(predicate)
 .reduce(_ && _)

但这将需要完整的集合来迭代，这是一种矫枉过正的做法。

我尝试了另一种适用于local[1]的方法，但似乎也遍历了真实集群上的所有内容：

rdd
 .map(predicate)
 .first()

[如果找不到任何需要的，则异常失败]

实现这一目标的规范方法是什么？

Answer 1

您可以使用aggregate：

def forAll[T](rdd:RDD[T])(p:T => Boolean): Boolean = {
  rdd.aggregate(true)((b, t) => b && p(t), _ && _)
}

作为旁注，没有真正的方法可以提前终止 spark，您将作业发送到集群并执行。聚合只是做你想做的事的好方法。