Enabling for Conversion to/from Pandas in Python
Arrow is available as an optimization when converting a Spark DataFrame to a Pandas DataFrame using the call toPandas() and when creating a Spark DataFrame from a Pandas DataFrame with createDataFrame(pandas_df). To use Arrow when executing these calls, users need to first set the Spark configuration spark.sql.execution.arrow.enabled to true. This is disabled by default.
In addition, optimizations enabled by
1
spark.sql.execution.arrow.enabled
Copied!
could fallback automatically to non-Arrow optimization implementation if an error occurs before the actual computation within Spark. This can be controlled by
1
spark.sql.execution.arrow.fallback.enabled.
Copied!
Example Python code
1
import findspark
2
findspark.init()
3
import pandas as pd
4
​
5
from pyspark.sql.functions import col, pandas_udf
6
from pyspark.sql.types import LongType
7
from pyspark.sql import SparkSession
8
​
9
spark = SparkSession.builder.getOrCreate()
10
​
11
# Declare the function and create the UDF
12
def multiply_func(a, b):
13
return a * b
14
​
15
multiply = pandas_udf(multiply_func, returnType=LongType())
16
​
17
# The function for a pandas_udf should be able to execute with local Pandas data
18
x = pd.Series([1, 2, 3])
19
print(multiply_func(x, x))
20
# 0 1
21
# 1 4
22
# 2 9
23
# dtype: int64
24
​
25
# Create a Spark DataFrame, 'spark' is an existing SparkSession
26
​
27
df = spark.createDataFrame(pd.DataFrame(x, columns=["x"]))
28
​
29
# Execute function as a Spark vectorized UDF
30
df.select(col("x")*col("x")).show()
31
​
32
'''
33
0 1
34
1 4
35
2 9
36
dtype: int64
37
+-------+
38
|(x * x)|
39
+-------+
40
| 1|
41
| 4|
42
| 9|
43
+-------+
44
'''
45
​
46
​
Copied!
Some issue:
1
import numpy as np
2
import pandas as pd
3
​
4
# Enable Arrow-based columnar data transfers
5
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
6
​
7
# Generate a Pandas DataFrame
8
pdf = pd.DataFrame(np.random.rand(100, 3))
9
​
10
# Create a Spark DataFrame from a Pandas DataFrame using Arrow
11
df = spark.createDataFrame(pdf)
12
​
13
# Convert the Spark DataFrame back to a Pandas DataFrame using Arrow
14
result_pdf = df.select("*").toPandas()
Copied!
here is the error when running creaeDataframe from pandas dataframe, when spark.sql.execution.arrow.enabled is true
1
/home/dv6/spark/spark/python/pyspark/sql/session.py:714: UserWarning: createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.enabled' is set to true; however, failed by the reason below:
2
An error occurred while calling z:org.apache.spark.sql.api.python.PythonSQLUtils.readArrowStreamFromFile.
3
: java.lang.IllegalArgumentException
4
at java.nio.ByteBuffer.allocate(ByteBuffer.java:334)
5
at org.apache.arrow.vector.ipc.message.MessageSerializer.readMessage(MessageSerializer.java:543)
6
at org.apache.spark.sql.execution.arrow.ArrowConverters$anon$3.readNextBatch(ArrowConverters.scala:243)
7
at org.apache.spark.sql.execution.arrow.ArrowConverters$anon$3.<init>(ArrowConverters.scala:229)
8
at org.apache.spark.sql.execution.arrow.ArrowConverters$.getBatchesFromStream(ArrowConverters.scala:228)
9
at org.apache.spark.sql.execution.arrow.ArrowConverters$anonfun$readArrowStreamFromFile$2.apply(ArrowConverters.scala:216)
10
at org.apache.spark.sql.execution.arrow.ArrowConverters$anonfun$readArrowStreamFromFile$2.apply(ArrowConverters.scala:214)
11
at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:2543)
12
at org.apache.spark.sql.execution.arrow.ArrowConverters$.readArrowStreamFromFile(ArrowConverters.scala:214)
13
at org.apache.spark.sql.api.python.PythonSQLUtils$.readArrowStreamFromFile(PythonSQLUtils.scala:46)
14
at org.apache.spark.sql.api.python.PythonSQLUtils.readArrowStreamFromFile(PythonSQLUtils.scala)
15
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
16
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
17
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
18
at java.lang.reflect.Method.invoke(Method.java:498)
19
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
20
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
21
at py4j.Gateway.invoke(Gateway.java:282)
22
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
23
at py4j.commands.CallCommand.execute(CallCommand.java:79)
24
at py4j.GatewayConnection.run(GatewayConnection.java:238)
25
at java.lang.Thread.run(Thread.java:748)
26
​
27
Attempting non-optimization as 'spark.sql.execution.arrow.fallback.enabled' is set to true.
28
warnings.warn(msg)
Copied!
Work around, set OS environment variable
1
export ARROW_PRE_0_15_IPC_FORMAT=1
Copied!
Then run Python code
1
(spark) [email protected]:~$ export ARROW_PRE_0_15_IPC_FORMAT=1
2
(spark) [email protected]:~$ python
3
Python 3.6.10 |Anaconda, Inc.| (default, Jan 7 2020, 21:14:29)
4
[GCC 7.3.0] on linux
5
Type "help", "copyright", "credits" or "license" for more information.
6
>>> import findspark
7
>>> findspark.init()
8
>>> import pandas as pd
9
>>>
10
>>> from pyspark.sql.functions import col, pandas_udf
11
>>> from pyspark.sql.types import LongType
12
>>> from pyspark.sql import SparkSession
13
>>>
14
>>> spark = SparkSession.builder.getOrCreate()
15
20/04/12 12:29:18 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
16
Setting default log level to "WARN".
17
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
18
20/04/12 12:29:20 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
19
20/04/12 12:29:20 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.
20
20/04/12 12:29:20 WARN Utils: Service 'SparkUI' could not bind on port 4042. Attempting port 4043.
21
20/04/12 12:29:20 WARN Utils: Service 'SparkUI' could not bind on port 4043. Attempting port 4044.
22
>>> import numpy as np
23
>>> import pandas as pd
24
>>>
25
>>> # Enable Arrow-based columnar data transfers
26
... spark.conf.set("spark.sql.execution.arrow.enabled", "true")
27
>>>
28
>>> # Generate a Pandas DataFrame
29
... pdf = pd.DataFrame(np.random.rand(100, 3))
30
>>> pdf
31
0 1 2
32
0 0.937892 0.387147 0.590136
33
1 0.007276 0.961907 0.156945
34
2 0.212474 0.048127 0.936995
35
3 0.074513 0.579899 0.803862
36
4 0.324786 0.352669 0.602877
37
.. ... ... ...
38
95 0.164290 0.376453 0.388663
39
96 0.014815 0.709746 0.615609
40
97 0.797867 0.563372 0.132668
41
98 0.755495 0.589192 0.793425
42
99 0.505420 0.672960 0.452064
43
​
44
[100 rows x 3 columns]
45
>>> df = spark.createDataFrame(pdf)
46
>>> # Convert the Spark DataFrame back to a Pandas DataFrame using Arrow
47
... result_pdf = df.select("*").toPandas()
48
>>>
49
​
Copied!
​
Last modified 1yr ago
Copy link