Wie funktioniert die pyspark mapPartitions-Funktion?

Question

Also versuche ich, Spark mit Python (Pyspark) zu lernen. Ich möchte wissen, wie die Funktion mapPartitions funktioniert. Das ist es, was Input benötigt und was Output gibt. Ich konnte kein richtiges Beispiel aus dem Internet finden. Nehmen wir an, ich habe ein RDD-Objekt mit Listen, wie unten.

[ [1, 2, 3], [3, 2, 4], [5, 2, 7] ] 

Und ich mag Element 2 aus allen Listen entfernen, wie würde ich erreichen, dass mapPartitions verwenden.

Answer 1

mapPartition sollte als eine Kartenoperation über Partitionen und nicht über die Elemente der Partition betrachtet werden. Die Eingabe ist die Menge der aktuellen Partitionen, deren Ausgabe eine weitere Gruppe von Partitionen sein wird.

Die Funktion, die Sie Karte passieren ein einzelnes Element des RDD nehmen muss

Die Funktion übergeben Sie mapPartition eine iterable Ihrer RDD Typen festgelegt werden und zurückkehren und iterable eines anderen oder gleichen Typs.

In Ihrem Fall Sie wollen wahrscheinlich nur etwas zu tun, wie

def filterOut2(line): 
    return [x for x in line if x != 2] 

filtered_lists = data.map(filterOut2) 

, wenn Sie es

def filterOut2FromPartion(list_of_lists): 
    final_iterator = [] 
    for sub_list in list_of_lists: 
    final_iterator.append([x for x in sub_list if x != 2]) 
    return iter(final_iterator) 

filtered_lists = data.mapPartition(filterOut2FromPartion) 

Answer 2

Es ist einfacher zu benutzen mapPartitions mit einer Generatorfunktion mit der yield wäre verwenden mapPartition wollte Syntax:

def filter_out_2(partition): 
    for element in partition: 
     if element != 2: 
      yield element 

filtered_lists = data.mapPartition(filter_out_2)