Spark RDD - 使用额外参数进行映射

Question

是否可以将额外的参数传递给pySpark中的映射函数？具体来说,我有以下代码配方:

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)

processDataLine除了JSON对象之外,该函数还需要额外的参数,如下所示:

def processDataLine(dataline, arg1, arg2)

如何传递额外的参数arg1,并arg2在flaMap功能？

Answer 1

1. 您可以直接在a中使用匿名函数 flatMap

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

   或者咖喱 processDataLine

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

2. 你可以processDataLine像这样生成:

   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库提供有用的curry装饰:

   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))
   

   请注意,我已将dataline参数推到最后一个位置.这不是必需的,但这样我们就不必使用关键字args.

4. 最后functools.partial,Avihoo Mamka在评论中已经提到过.