Pandas 标量 UDF 失败，IllegalArgumentException

Question

首先，如果我的问题很简单，我深表歉意。我确实花了很多时间研究它。

我正在尝试按照此处所述在PySpark 脚本中设置标量 Pandas UDF。

这是我的代码：

from pyspark import SparkContext
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql import SQLContext
sc.install_pypi_package("pandas")
import pandas as pd
sc.install_pypi_package("PyArrow")

df = spark.createDataFrame(
    [("a", 1, 0), ("a", -1, 42), ("b", 3, -1), ("b", 10, -2)],
    ("key", "value1", "value2")
)

df.show()

@F.pandas_udf("double", F.PandasUDFType.SCALAR)
def pandas_plus_one(v):
    return pd.Series(v + 1)

df.select(pandas_plus_one(df.value1)).show()
# Also fails
#df.select(pandas_plus_one(df["value1"])).show()
#df.select(pandas_plus_one("value1")).show()
#df.select(pandas_plus_one(F.col("value1"))).show()

脚本在最后一条语句失败：

调用 o209.showString 时出错。：org.apache.spark.SparkException：作业因阶段失败而中止：阶段 8.0 中的任务 2 失败 4 次，最近失败：阶段 8.0 中丢失任务 2.3（TID 30，ip-10-160-2-53.ec2 .内部，执行者 3)： java.lang.IllegalArgumentException 在 java.nio.ByteBuffer.allocate(ByteBuffer.java:334) 在 org.apache.arrow.vector.ipc.message.MessageSerializer.readMessage(MessageSerializer.java:543) 在 org.apache.arrow.vector.ipc.message .MessageChannelReader.readNext(MessageChannelReader.java:58) 在 org.apache.arrow.vector.ipc.ArrowStreamReader.readSchema(ArrowStreamReader.java:132) 在 org.apache.arrow.vector.ipc.ArrowReader.initialize(ArrowReader.java) :181) 在 org.apache.arrow.vector.ipc.ArrowReader.ensureInitialized(ArrowReader.java:172) 在 org.apache.arrow.vector.ipc.ArrowReader.getVectorSchemaRoot(ArrowReader.java:65) 在 org.apache。 spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:162) 在 org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:122) 在org.apache。spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:410) ...

我在这里缺少什么？我只是按照手册。谢谢你的帮助

Answer 1

Pyarrow 在 2019 年 10 月 5 日推出了一个新版本 0.15，这导致 pandas Udf 抛出错误。Spark 需要升级才能与此兼容（这可能需要一些时间）。您可以在此处关注进度https://issues.apache.org/jira/projects/SPARK/issues/SPARK-29367?filter=allissues

解决方案：

1. 您需要安装 Pyarrow 0.14.1 或更低版本。< sc.install_pypi_package("pyarrow==0.14.1") > (或)
2. 从使用 Python 的位置设置环境变量ARROW_PRE_0_15_IPC_FORMAT=1