在星火RDD

Question

我有一個字符串的一大RDD懶惰的foreach（通過幾種sc.textFile(...))工會獲得。

我現在想搜索該RDD給定的字符串，我想搜索時停止一個「足夠好」的比賽已經找到。

我可以改造foreach，或filter，或map用於此目的，但所有這些都將通過每一個元素在RDD迭代，不管比賽是否已經達到。

有沒有辦法將這個過程短路並避免id遍歷整個RDD？

Answer 1

我可以改造的foreach，或過濾器，或映射用於此目的，但所有這些都將通過每一個元素在RDD

其實，你錯了迭代。星火引擎是足夠聰明的優化計算，如果你限制的結果（利用take或first）：

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() 

現在，讓我們檢查ACCUM：

>>> 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 function for huge lists。

Answer 2

不是。沒有find方法，就像啓發Spark API的Scala集合一樣，一旦找到滿足謂詞的元素就會停止查找。大概你最好的選擇是使用一個數據源，它可以最大限度地減少多餘的掃描，比如Cassandra，驅動程序會在這裏下推一些查詢參數。你也可以看看更實驗的名爲BlinkDB的伯克利項目。底線，Spark更多地用於掃描數據集，比如之前的MapReduce，而不是傳統的類似於數據庫的查詢。