I am using pandas_udf of type PandasUDFType.GROUPED_MAP which returns Pandas dataframe containing each column as dtype object ( in actual an array object ) for each column in input pyspark dataframe.
Pyspark input dataframe looks like this:-
id |c1 | c2 | c3 | c4 |
-----------------------
1 |0 | 1 | 2 | 3 |
2 |9 | 6 | 8 | 4 |
Here is a sample code:-
schema = StructType(List(StructField(c1,ArrayType(IntegerType(),true),true),StructField(c2,ArrayType(IntegerType(),true),true),StructField(c3,ArrayType(IntegerType(),true)
@pandas_udf(schema,PandasUDFType.GROUPED_MAP)
def convert_to_array_udf(pdf):
//logic to convert scalar value to array as illustrated above
print(outputdf.head(2)
print(outputdf.dtypes)
return outputdf
pyspark_output_df = input_df.groupyby("id").apply(convert_to_array_udf)
output of the execution is:-
c1 | c2 | c3 | c4 |
-----------------------------------
[0,0] | [1,1] | [2,2] | [3,3] |
[9,9] | [6,6] | [8,8] | [4,4] |
c1 object
c2 object
c3 object
dtype: object
File "project\venv\Lib\site-packages\pyspark\python\lib\pyspark.zip\pyspark\serializers.py", line 253, in create_array
return pa.Array.from_pandas(s, mask=mask, type=t)
File "array.pxi", line 335, in pyarrow.lib.Array.from_pandas
File "array.pxi", line 170, in pyarrow.lib.array
File "array.pxi", line 70, in pyarrow.lib._ndarray_to_array
File "error.pxi", line 77, in pyarrow.lib.check_status
pyarrow.lib.ArrowInvalid: Error inferring Arrow data type for collection of Python objects. Got Python object of type int64 but can only handle these types: bool, float, integer, date, datetime, bytes, unicode
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:452)
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:172)
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:122)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:406)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
I am using python 3.6.8, pyspark 2.4.3, pyarrow 5.0.0 (although also tested in pyarrow 0.8.0) and pandas 1.1.5. Does pyarrow not support arrays as of now?