Spark __getnewargs__错误

我试图通过将它映射到RDD然后返回到DataFrame来清理Spark DataFrame.这是一个玩具示例:

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

运行上面的代码会产生Py4JError一个非常长的堆栈跟踪,结尾如下:

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)

这里发生了什么？我如何解决它？我正在使用PySpark 1.5.2.

1> Paul..：

---

该错误是由参考引起ex.columns的.map(lambda...)声明.您不能在RDD转换中使用的函数内引用RDD.在这种情况下,Spark 应该会发出更多有用的错误,但显然没有进入这个版本.

解决方案是使用引用变量的副本替换引用:

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

---

你说的直接原因是正确的,但它与你所链接的JIRA不同.忽略酸洗问题,这与我在[这里](http://stackoverflow.com/q/31684842/1560062)描述的问题几乎相同.由于DataFrame只是JVM对象的包装器,因此无法从PySpark操作/转换中访问它.但它在Scala中运行得很好.