Spark RDD-Mapeo con argumentos adicionales

Question

Spark RDD-Mapeo con argumentos adicionales

¿Es posible pasar argumentos adicionales a la función de asignación en pySpark? Específicamente, tengo la siguiente receta de código:

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)

La función processDataLine toma argumentos adicionales además del objeto JSON, como:

def processDataLine(dataline, arg1, arg2)

¿Cómo puedo pasar los argumentos adicionales arg1 y arg2 a la función flaMap?

26

python apache-spark rdd pyspark

Author: zero323, 2015-10-08

Source

1 answers

score 36 · Accepted Answer

Puede usar una función anónima directamente en un flatMap

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

O al curry processDataLine

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

Puedes generar processDataLine así:

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

json_data_rdd.flatMap(processDataLine(arg1, arg2))

toolz biblioteca proporciona útil curry decorador:
```
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))
```
Tenga en cuenta que he empujado dataline argumento a la última posición. No es necesario, pero de esta manera no tenemos que usar args de palabras clave.
Finalmente hay functools.partial ya mencionado por Avihoo Mamka en los comentarios.