Spark RDD上的懒惰foreach

Question

我有很大的字符串RDD（通过几个的联合获得）sc.textFile(...))。

现在，我想在该RDD中搜索给定的字符串，并且当找到“足够好”的匹配项时，我希望搜索停止。

我可以为此进行改进foreach，或filter，map但是所有这些都将遍历该RDD中的每个元素，而不管是否已达到匹配条件。

有没有办法缩短该过程并避免迭代整个RDD？

Answer 1

我可以为此目的改进foreach，或过滤或映射，但是所有这些都会迭代该RDD中的每个元素

其实你错了 如果您限制结果（使用take或first），Spark引擎足够聪明，可以优化计算：

import numpy as np
from __future__ import print_function

np.random.seed(323)

acc = sc.accumulator(0)

def good_enough(x, threshold=7000):
    global acc
    acc += 1
    return x > threshold

rdd = sc.parallelize(np.random.randint(0, 10000) for i in xrange(1000000))

x = rdd.filter(good_enough).first()

现在让我们检查一下累加：

>>> print("Checked {0} items, found {1}".format(acc.value, x))
Checked 6 items, found 7109

并确保一切都按预期进行：

acc = sc.accumulator(0)
rdd.filter(lambda x: good_enough(x, 100000)).take(1)
assert acc.value == rdd.count()

可以使用数据帧和udf以更有效的方式完成相同的操作。

注意：在某些情况下，甚至可以在Spark中使用无限序列，但仍然会得到结果。您可以查看我对Spark FlatMap函数的答案以获取大量示例。