Come funziona la funzione mapPartitions di pyspark?

Question

Quindi sto cercando di imparare Spark usando Python (Pyspark). Voglio sapere come funziona la funzione mapPartitions. Questo è ciò che Input prende e quale Output dà. Non sono riuscito a trovare alcun esempio corretto da internet. Diciamo, ho un oggetto RDD contenente elenchi, come di seguito.

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

E voglio rimuovere l'elemento 2 da tutte le liste, come faccio a ottenere che l'utilizzo di mapPartitions.

Answer 1

mapPartition deve essere considerato un'operazione di mapping su partizioni e non sugli elementi della partizione. È l'input è l'insieme delle partizioni correnti, il suo output sarà un altro insieme di partizioni.

La funzione si passa mappa deve prendere un singolo elemento della vostra RDD

La funzione si passa mapPartition deve prendere un iterabile del vostro tipo di RDD e tornare e iterabile di qualche altro o lo stesso tipo.

Nel tuo caso, probabilmente si vuole solo fare qualcosa di simile

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

filtered_lists = data.map(filterOut2) 

se si voleva utilizzare mapPartition sarebbe

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

E 'più facile da usare mapPartitions con una funzione di generatore utilizzando il yield sintassi:

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

filtered_lists = data.mapPartition(filter_out_2)