Jak działa funkcja Pyspark mapPartitions?

Question

Próbuję nauczyć się Sparka używając Pythona (Pyspark). Chcę wiedzieć, jak działa ta funkcja. To, co Input bierze i jaki wynik daje. Nie mogłem znaleźć odpowiedniego przykładu z internetu. Powiedzmy, mam obiekt RDD zawierający listy, takie jak poniżej.

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

I chcę, aby usunąć element 2 ze wszystkich list, jaki chciałbym osiągnąć stosując mapPartitions.

Answer 1

mapPartycja powinna być traktowana jako operacja mapowa na partycjach, a nie na elementach partycji. Jego wejście jest zbiorem aktualnych partycji, a jego wyjście będzie kolejnym zbiorem partycji.

Funkcja zdać mapa musi mieć indywidualny element swojej RDD

funkcji, którą przechodzą mapPartition musi brać iterowalny swojego rodzaju RDD i powrócić i iterable jakiegoś innego lub tego samego typu.

W twoim przypadku prawdopodobnie po prostu chcesz zrobić coś takiego

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

filtered_lists = data.map(filterOut2) 

jeśli chcesz skorzystać mapPartition byłoby

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

Łatwiej jest użyć mapPartitions z funkcją generatora pomocą yield składnia:

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

filtered_lists = data.mapPartition(filter_out_2)