pyspark mapPartitions işlevi nasıl çalışır?

Question

Bu yüzden Spark'i Python (Pyspark) kullanarak öğrenmeye çalışıyorum. mapPartitions işlevinin nasıl çalıştığını bilmek istiyorum. Girdiğin şey budur ve ne çıktı verir. İnternetten uygun bir örnek bulamadım. Diyelim ki, aşağıdaki gibi bir liste içeren bir RDD nesnesine sahibim.

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

Ve ben tüm listelerden eleman 2 kaldırmak istediğiniz, nasıl mapPartitions kullanarak Bunu başarmak olacaktır.

Answer 1

mapPartition, bölümlerin öğeleri üzerinde değil, bölümler üzerinde bir harita işlemi olarak düşünülmelidir. Giriş, mevcut bölümlerin kümesidir, çıktısı başka bir bölüm kümesi olacaktır.

Eğer harita mapPartition senin RDD türünde bir iterable almalıdır geçmesi için RDD

fonksiyonu tek bir öğesine almak ve diğer bazı veya aynı tipte dönmek ve iterable gerekir geçmesi işlevi. Senin durumunda

muhtemelen sadece bu yield kullanarak bir jeneratör işlevi ile mapPartitions kullanımı daha kolay

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

olurdu Eğer mapPartition kullanmak istiyorsa

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

filtered_lists = data.map(filterOut2) 

gibi bir şey yapmak istiyorum sözdizimi:

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

filtered_lists = data.mapPartition(filter_out_2)