火花 __getnewargs__ 错误答案

【问题标题】：Spark __getnewargs__ error火花 __getnewargs__ 错误
【发布时间】：2016-03-30 08:22:10
【问题描述】：

我正在尝试通过将 Spark DataFrame 映射到 RDD 然后返回 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。

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

该错误是由.map(lambda...) 语句中对ex.columns 的引用引起的。您不能在 RDD 转换中使用的函数中引用 RDD。 Spark 是 supposed to issue more helpful errors in this case，但显然它没有进入这个版本。

解决方案是用被引用变量的副本替换引用：

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 是不同的问题。忽略酸洗问题，这与我描述的here 几乎相同。由于 DataFrame 只是 JVM 对象的包装器，因此无法从 PySpark 操作/转换中访问它。但它在 Scala 中可以正常工作。