pandas - Python pandas_udf Spark 错误

Question

我开始在本地玩 Spark 并发现这个奇怪的问题

1) pip install pyspark==2.3.1
2)pyspark>

将 Pandas 导入为 pd
从 pyspark.sql.functions 导入 pandas_udf, PandasUDFType, udf
df = pd.DataFrame({'x': [1,2,3], 'y':[1.0,2.0,3.0]})
sp_df = spark.createDataFrame(df)

@pandas_udf('long', PandasUDFType.SCALAR)
def pandas_plus_one(v):
返回 v + 1

sp_df.withColumn('v2', pandas_plus_one(sp_df.x)).show()

以这里为例 https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html

知道为什么我不断收到此错误吗？

py4j.protocol.Py4JJavaError:调用 o108.showString 时出错。
:org.apache.spark.SparkException:由于阶段失败而中止作业:阶段 3.0 中的任务 0 失败 1 次，最近失败:阶段 3.0 中丢失任务 0.0(TID 8、本地主机、执行程序驱动程序):org.apache.spark .SparkException: Python worker 意外退出(崩溃)
在 org.apache.spark.api.python.BasePythonRunner$ReaderIterator$$anonfun$1.applyOrElse(PythonRunner.scala:333)
在 org.apache.spark.api.python.BasePythonRunner$ReaderIterator$$anonfun$1.applyOrElse(PythonRunner.scala:322)
在 scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
在 org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:177)
在 org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:121)
在 org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
在 org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
在 org.apache.spark.sql.execution.python.ArrowEvalPythonExec$$anon$2.(ArrowEvalPythonExec.scala:90)
在 org.apache.spark.sql.execution.python.ArrowEvalPythonExec.evaluate(ArrowEvalPythonExec.scala:88)
在 org.apache.spark.sql.execution.python.EvalPythonExec$$anonfun$doExecute$1.apply(EvalPythonExec.scala:131)
在 org.apache.spark.sql.execution.python.EvalPythonExec$$anonfun$doExecute$1.apply(EvalPythonExec.scala:93)
在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:800)
在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:800)
在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
在 org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
在 org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
在 org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
在 org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
在 org.apache.spark.scheduler.Task.run(Task.scala:109)
在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
在 java.lang.Thread.run(Thread.java:745)
引起:java.io.EOFException
在 java.io.DataInputStream.readInt(DataInputStream.java:392)
在 org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:158)
……还有 27 个

Answer 1

我有同样的问题。我发现这是 pandas 和 numpy 之间的版本问题。

对我来说，以下工作:

numpy==1.14.5
pandas==0.23.4
pyarrow==0.10.0

在我有以下非工作组合之前:

numpy==1.15.1
pandas==0.23.4
pyarrow==0.10.0