在另一個RDD的基礎上修剪一個RDD

Question

在另一個RDD的基礎上，是否有一種方法可以在一個RDD中篩選元素，即使它們不共享相同的密鑰？

我有兩個RDDS - ABC和XYZ

abc.collect（）看起來像這樣

[[a,b,c],[a,c,g,e],[a,e,b,x],[b,c]] 

xyz.collect（）看起來像這樣

[a,b,c] 

現在我想濾除RDD abc中所有不在xyz中的元素。

上述操作後，RDD美國廣播公司應該是這樣的：

[[a,b,c],[a,c],[a,b],[b,c]] 

我寫的，看起來像這樣的代碼：

​​

然而，這給了我這個錯誤：

Exception: It appears that you are attempting to broadcast an RDD or reference an RDD from an action or transformation. RDD transformations and actions can only be invoked by the driver, not inside of other transformations; for example, rdd1.map(lambda x: rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation 

我已經嘗試過濾器，查找而不是地圖無濟於事。我不斷收到相同的錯誤。

我知道我可以對xyz執行收集操作並解決此錯誤，但是我在大型數據集上運行此操作，並且執行.collect（）會導致我的AWS服務器超出太多內存。因此，我需要在不使用.collect（）或任何此類等效的昂貴操作的情況下執行此操作。

Answer 1

您可以：

# Add index 
abc.zipWithIndex() \ 
    # Flatten values 
    .flatMap(lambda x: [(k, x[1]) for k in x[0]]) \ 
    # Join with xyz (works as filter) 
    .join(xyz.map(lambda x: (x, None))) \ 
    # Group back by index 
    .map(lambda x: (x[1][0], x[0])) \ 
    .groupByKey() \ 
    .map(lambda x: list(x[1])) 

，或者可以創建布隆過濾器上xyz並用它來繪製abc。