Spark __getnewargs__ error

Estoy intentando limpiar un Spark DataFrame asignándolo a RDD y luego a DataFrame. Aquí hay un ejemplo de juguete:

def replace_values(row,sub_rules): d = row.asDict() for col,old_val,new_val in sub_rules: if d[col] == old_val: d[col] = new_val return Row(**d) ex = sc.parallelize([{'name': 'Alice', 'age': 1},{'name': 'Bob', 'age': 2}]) ex = sqlContext.createDataFrame(ex) (ex.map(lambda row: replace_values(row,[(col,1,3) for col in ex.columns])) .toDF(schema=ex.schema)) 

La ejecución del código anterior da como resultado un Py4JError con un seguimiento de stack muy largo que termina en lo siguiente:

 Py4JError: An error occurred while calling o801.__getnewargs__. Trace: py4j.Py4JException: Method __getnewargs__([]) does not exist at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:333) at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:342) at py4j.Gateway.invoke(Gateway.java:252) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:207) at java.lang.Thread.run(Thread.java:745) 

¿Que esta pasando aqui? ¿Cómo lo arreglo? Estoy usando PySpark 1.5.2.

El error se debe a la referencia a ex.columns en la statement .map(lambda...) . No puede tener referencias a un RDD dentro de la función que se utiliza en una transformación RDD. Se supone que Spark emite errores más útiles en este caso , pero aparentemente eso no lo hizo en esta versión.

La solución es reemplazar las referencias con copias de las variables referenciadas:

 def replace_values(row,sub_rules): d = row.asDict() for col,old_val,new_val in sub_rules: if d[col] == old_val: d[col] = new_val return Row(**d) ex = sc.parallelize([{'name': 'Alice', 'age': 1},{'name': 'Bob', 'age': 2}]) ex = sqlContext.createDataFrame(ex) cols = copy.deepcopy(ex.columns) (ex.map(lambda row: replace_values(row,[(col,1,3) for col in cols])) .toDF(schema=ex.schema))