Eine RDD auf der Basis einer anderen RDD beschneiden

Question

Gibt es eine Möglichkeit, Filterelemente in einer RDD auf der Basis einer anderen RDD zu filtern, auch wenn sie nicht den gleichen Schlüssel haben?

Ich habe zwei RDDs - abc und xyz

abc.collect() sieht wie folgt aus

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

xyz.collect() dieses Jetzt wollen

[a,b,c] 

sieht aus wie ich filter out alle Elemente von RDD abc, die nicht in xyz vorhanden sind.

Nach dem genannten Betrieb RDD Abc sollte wie folgt aussehen:

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

ich einen Code geschrieben, die etwa wie folgt aussieht:

def prune(eachlist): 
    for i in eachlist: 
     if i in xyz: 
      return i 

abc = abc.map(prune) 

jedoch, dass mir diesen Fehler gibt:

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 

Ich habe seit dem versucht, Filter, Lookups statt Karte ohne Erfolg. Ich bekomme immer den gleichen Fehler.

Ich weiß, dass ich eine Sammlungsoperation auf xyz durchführen und diesen Fehler beheben kann, aber ich führe dies auf einem großen Dataset aus, und eine .collect() führt meinen AWS-Server zum Überschreiten zu viel Arbeitsspeicher. Daher muss ich dies tun, ohne .collect() oder ähnliche äquivalente teure Operationen zu verwenden.

Answer 1

Sie können:

# 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])) 

oder Sie können Bloom auf xyz Filter erstellen und verwenden Sie es abc abzubilden.