调用Koalas DataFrame的dfq.head()出现ValueError与SocketTimeoutException问题

问题场景 #

对Koalas DataFrame执行dfq.head()时出现错误，确认与数据本身无关，由库版本不兼容导致。

初始化代码 #

spark = (SparkSession
         .builder
         .appName("my_app")
         .config('spark.yarn.appMasterEnv.ARROW_PRE_0_15_IPC_FORMAT', 1)
         .config('spark.executorEnv.ARROW_PRE_0_15_IPC_FORMAT', 1)
         .config('spark.yarn.appMasterEnv.PYARROW_IGNORE_TIMEZONE', 1)
         .config('spark.executorEnv.PYARROW_IGNORE_TIMEZONE', 1)
         .enableHiveSupport()
         .getOrCreate())

df = spark.read.table('sometable')

import databricks.koalas as koalas
dfq = df.to_koalas()
dfq.head()

报错信息 #

~/.local/lib/python3.7/site-packages/databricks/koalas/internal.py in to_pandas_frame(self)
    932         """ Return as pandas DataFrame. """
    933         sdf = self.to_internal_spark_frame
--> 934         pdf = sdf.toPandas()
    935         if len(pdf) == 0 and len(sdf.schema) > 0:
    936             pdf = pdf.astype(

~/.local/lib/python3.7/site-packages/pyspark/sql/dataframe.py in toPandas(self)
   2127                         _check_dataframe_localize_timestamps
   2128                     import pyarrow
--> 2129                     batches = self._collectAsArrow()
   2130                     if len(batches) > 0:
   2131                         table = pyarrow.Table.from_batches(batches)

~/.local/lib/python3.7/site-packages/pyspark/sql/dataframe.py in _collectAsArrow(self)
   2185         with SCCallSiteSync(self._sc):
   2186             from pyspark.rdd import _load_from_socket
--> 2187             port, auth_secret, jsocket_auth_server = self._jdf.collectAsArrowToPython()
   2188             try:
   2189                 return list(_load_from_socket((port, auth_secret), ArrowStreamSerializer()))

ValueError: not enough values to unpack (expected 3, got 2)
Exception in thread "serve-Arrow" java.net.SocketTimeoutException: Accept timed out
    at java.net.PlainSocketImpl.socketAccept(Native Method)
    at java.net.AbstractPlainSocketImpl.accept(AbstractPlainSocketImpl.java:409)
    at java.net.ServerSocket.implAccept(ServerSocket.java:560)
    at java.net.ServerSocket.accept(ServerSocket.java:528)
    at org.apache.spark.api.python.PythonServer$$anon$1.run(PythonRDD.scala:883)
Exception in thread "serve-Arrow" java.net.SocketTimeoutException: Accept timed out
    at java.net.PlainSocketImpl.socketAccept(Native Method)
    at java.net.AbstractPlainSocketImpl.accept(AbstractPlainSocketImpl.java:409)
    at java.net.ServerSocket.implAccept(ServerSocket.java:560)
    at java.net.ServerSocket.accept(ServerSocket.java:528)
    at org.apache.spark.api.python.PythonServer$$anon$1.run(PythonRDD.scala:883)

解决方案 #

这个错误是PySpark、PyArrow、Databricks Koalas版本不匹配导致的，具体解决步骤如下：

• 版本匹配规则：

  • Spark 2.4.x 搭配 PyArrow 0.14.x 及更低版本，Koalas 对应 0.24.x 系列
  • Spark 3.x 需搭配官方推荐的PyArrow版本（如Spark 3.0→PyArrow 0.15.x，Spark 3.1→PyArrow 0.16.x，Spark 3.2→PyArrow 6.x），Koalas 对应 1.x 系列

• 调整Spark配置：
  移除针对旧版PyArrow的ARROW_PRE_0_15_IPC_FORMAT配置，仅保留时区忽略配置即可：

  spark = (SparkSession
           .builder
           .appName("my_app")
           .config('spark.yarn.appMasterEnv.PYARROW_IGNORE_TIMEZONE', 1)
           .config('spark.executorEnv.PYARROW_IGNORE_TIMEZONE', 1)
           .enableHiveSupport()
           .getOrCreate())
  

• 重装匹配版本的库：
  以Spark 3.1为例，执行以下命令安装对应版本：

  pip install pyarrow==0.16.0 databricks-koalas==1.8.0
  

内容的提问来源于stack exchange，提问作者heinistic