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)

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

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检查中定义然后返回。