Spark RDD - Mapowanie z dodatkowymi argumentami

Question

Czy jest możliwe przekazanie dodatkowych argumentów do funkcji mapowania w pySpark? Konkretnie mam kodu następujący przepis:

raw_data_rdd = sc.textFile("data.json", use_unicode=True) 
json_data_rdd = raw_data_rdd.map(lambda line: json.loads(line)) 
mapped_rdd = json_data_rdd.flatMap(processDataLine) 

Funkcja processDataLine wykonuje dodatkowe argumenty oprócz obiektu JSON, jak:

def processDataLine(dataline, arg1, arg2) 

jaki sposób można przekazać dodatkowe argumenty arg1 i arg2 do funkcja flaMap?

Answer 1

1. Można użyć anonimową funkcję bezpośrednio w flatMap

   json_data_rdd.flatMap(lambda j: processDataLine(j, arg1, arg2)) 
   

   lub curry processDataLine

   f = lambda j: processDataLine(dataline, arg1, arg2) 
   json_data_rdd.flatMap(f) 
   

2. Można wygenerować processDataLine takiego:

   def processDataLine(arg1, arg2): 
       def _processDataLine(dataline): 
        return ... # Do something with dataline, arg1, arg2 
       return _processDataLine 
   
   json_data_rdd.flatMap(processDataLine(arg1, arg2)) 
   

3. toolz biblioteka dostarcza użytecznych curry dekorator:

   from toolz.functoolz import curry 
   
   @curry 
   def processDataLine(arg1, arg2, dataline): 
       return ... # Do something with dataline, arg1, arg2 
   
   json_data_rdd.flatMap(processDataLine(arg1, arg2)) 
   

   Zauważ, że mam popychane dataline argumentu do ostatniej pozycji. Nie jest to wymagane, ale w ten sposób nie musimy używać argumentów słów kluczowych.

4. Wreszcie jest functools.partial wspomniana już przez Avihoo Mamka w komentarzach.