Pandas UDF 方法 getstate([]) 不存在错误

Question

我正在使用 pyspark 2.4.1 并尝试使用如下所示的 Pandas UDF 编写一个简单的函数。基本上创建一个新列并根据 df.x=='a' 和 df.y=='t' 分配字符串值。但是，我不断收到 Method __getstate__([]) does not exist 错误。这是我尝试使用 Pandas UDF 的 2 种方法，但不确定还有什么其他方法可以编写它：

数据

x = pd.Series(['a', 'b', 'c'])
y = pd.Series(['t','t','t'])

df = spark.createDataFrame(pd.DataFrame({"x":x,"y":y}))
df.show()

+---+---+
|  x|  y|
+---+---+
|  a|  t|
|  b|  t|
|  c|  t|
+---+---+

尝试 1：

from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql.types import StringType

import pandas as pd

@pandas_udf(StringType(), PandasUDFType.SCALAR)
def test_fun(x: str, y: str) -> pd.Series:
    import os
    os.environ["ARROW_PRE_0_15_IPC_FORMAT"] = "1"
    if x.values=='a' and y.values=='t':
        return z == 'ok'
    else:
        return z == "None"
    return pd.Series(z)

df.withColumn('test',test_fun(col("x"),col("y"))).show()

尝试 2

def test_func(df):
    @pandas_udf(StringType(), PandasUDFType.SCALAR)
    def test(x: str, y: str) -> pd.Series:
        import os
        os.environ["ARROW_PRE_0_15_IPC_FORMAT"] = "1"
        if x.values=='a' and y.values=='t':
            return z == 'ok'
        else:
            return z == "None"
        return pd.Series(z) 
     
    return df.withColumn('test', test(col('x'),col('y')))

test_func(df)

两者都给了我同样的错误信息：

...py4j.protocol.Py4JError: An error occurred while calling t.__getstate__. Trace:
py4j.Py4JException: Method __getstate__([]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
    at py4j.Gateway.invoke(Gateway.java:274)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)

我对 spark 很陌生，阅读了许多有类似问题的线程，但无法找出修改它的正确方法。

Answer 1

我试过尝试 1，我相信这也适用于尝试 2。

在你的udf中，你返回了3次以及返回的方式，之前没有定义z。

以下函数在 spark 2.4.7 和 pyarrow==0.13.0 上对我有效

@pandas_udf(StringType(), PandasUDFType.SCALAR)
def test_fun(x: str, y: str) -> pd.Series:
    if x.values=='a' and y.values=='t':
        z = 'ok'
    else:
        z = "None"
    return pd.Series(z)

z应该在if-else检查中定义然后返回。

Pandas UDF 方法 getstate([]) 不存在错误

Pandas UDF Method getstate([]) does not exist error

pandas

apache-spark

pyspark

pyarrow

Pandas UDF 方法 __getstate__([]) 不存在错误

Pandas UDF Method __getstate__([]) does not exist error

pandas

apache-spark

pyspark

pyarrow

Pandas UDF 方法 getstate([]) 不存在错误

Pandas UDF Method getstate([]) does not exist error