檢查RDD中是否存在值

我已經在python中編寫了一個可正常工作的Spark程序。檢查RDD中是否存在值

但是，它在內存消耗方面效率低下&我正試圖優化它。我在AWS EMR上運行它，EMR正在消耗太多內存的工作。

Lost executor 11 on ip-*****: Container killed by YARN for exceeding memory limits. 11.4 GB of 10.4 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

我相信這個內存問題是由於這樣的事實，我在幾種情況下收集我的RDDS（（），即使用.collect），因爲在後期階段，我需要測試，如果在列表中存在一定的價值是否由這些RDD製成。

所以，目前我的代碼看起來像這樣：

myrdd = data.map(lambda word: (word,1))  \ 
     .reduceByKey(lambda a,b: a+b) \ 
     .filter(lambda (a, b): b >= 5) \ 
     .map(lambda (a,b) : a)   \ 
     .collect()

，並在代碼

if word in myrdd: 
    mylist.append(word) 

myrdd2 = data2.map(lambda word: (word,1))  \ 
     .reduceByKey(lambda a,b: a+b) \ 
     .filter(lambda (a, b): b >= 5) \ 
     .map(lambda (a,b) : a)   \ 
     .collect() 

if word in myrdd2: 
    mylist2.append(word)

，然後我重複這種模式多次晚些時候。

有沒有辦法做到操作

if word in myrdd: 
    do something

不先收集RDD？

是否有像rdd.contains（）這樣的函數？

P.S：我沒有在內存中緩存任何東西。我的火花背景是這樣的：從紗線

jobName = "wordcount" 
sc = SparkContext(appName = jobName) 

...... 
...... 

sc.stop()

來源

2016-11-25 Piyush

不使用.collect （）它會把所有的數據傳給驅動程序，如果你有更大的數據集，會產生一個問題。使用myrdd2.foreachRDD並檢查值是否存在 – Backtrack

word = sc.broadcast（[w1，w2，w3]） valuepresent = myrdd.filter {lambda x：x in word}這樣做也是一種解決方法，我會認爲 – Backtrack

錯誤消息指出collect不是一個問題，因爲你的執行者（而不是驅動器）有記憶問題。

首先，嘗試按照錯誤消息建議並提升spark.yarn.executor.memoryOverhead - 在YARN上運行pyspark時，您可以告訴YARN爲python工作進程內存分配更大的容器。

接下來，看看執行者需要大量內存的操作。你使用reduceByKey，也許你可以增加分區的數量，使它們在內存使用方面更小。看看numPartitions參數：http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.reduceByKey

最後，如果你想檢查是否RDD包含了一些值，那麼僅僅通過這個值過濾和檢查，使用count或first，例如：

looking_for = "....." 
contains = rdd.filter(lambda a: a == looking_for).count() > 0

來源

2016-11-25 05:46:49 Mariusz

謝謝。有很多RDD會對執行者造成壓力嗎？例如。如果我做了像myrddalias = myrdd之類的東西，它是否會給內存帶來額外的壓力，或者那樣好嗎？ – Piyush

它只是複製引用，rdds本身不會被克隆 – Mariusz

問題是，looking_for是一個RDD，當我在另一個RDD上執行過濾器時，它向我顯示一個錯誤，說我不能將一個轉換放在另一個RDD中。 Looking_for是一個列表，我希望根據存在於look_for rdd中的某個值來修剪我的rdd。確切的錯誤 - 例外：看起來您正試圖廣播RDD或引用操作或轉換中的RDD。 RDD轉換和操作只能由驅動程序調用，而不能在其他轉換中調用; – Piyush

檢查RDD中是否存在值

回答

相關問題