Spark DataFrame调用toPandas()转Pandas时出现Socket连接报错

问题现象 #

Spark作业本身可正常执行完成，仅在将Spark DataFrame收集转换为Pandas DataFrame的环节抛出异常，调整内存、超时等多项配置后问题仍未解决。

现有作业提交配置 #

RUN_OPTIONS="--driver-memory 18g --executor-cores 2 --executor-memory 14g --conf spark.driver.maxResultSize=10g --conf spark.maxRemoteBlockSizeFetchToMem=2g --conf spark.executor.heartbeatInterval=3600s --conf spark.network.timeout=3610s --conf spark.worker.timeout=3610s --conf spark.dynamicAllocation.minExecutors=5 --conf spark.dynamicAllocation.maxExecutors=50 --conf spark.sql.broadcastTimeout=36000 "

核心实现代码 #

SPARK_DF=DF_RESULT.union(dict_dq[keys])
DF_PD=SPARK_DF.toPandas()

完整报错信息 #

Exception in thread "serve-DataFrame" java.net.SocketTimeoutException: Accept timed out
    at java.net.PlainSocketImpl.socketAccept(Native Method)
    at java.net.AbstractPlainSocketImpl.accept(AbstractPlainSocketImpl.java:535)
    at java.net.ServerSocket.implAccept(ServerSocket.java:545)
    at java.net.ServerSocket.accept(ServerSocket.java:513)
    at org.apache.spark.api.python.PythonServer$$anon$1.run(PythonRDD.scala:886)

Traceback (most recent call last):
  File "/opt/cgfiles/common/DQ_Check/main/runDQ.py", line 469, in <module>
    DF_RES_PD=DF_RESULT.toPandas()
  File "/opt/cloudera/parcels/CDH-6.2.1-1.cdh6.2.1.p5242.21315880/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 2142, in toPandas
  File "/opt/cloudera/parcels/CDH-6.2.1-1.cdh6.2.1.p5242.21315880/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 534, in collect
  File "/opt/cloudera/parcels/CDH-6.2.1-1.cdh6.2.1.p5242.21315880/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 144, in _load_from_socket
  File "/opt/cloudera/parcels/CDH-6.2.1-1.cdh6.2.1.p5242.21315880/lib/spark/python/lib/pyspark.zip/pyspark/java_gateway.py", line 178, in local_connect_and_auth
Exception: could not open socket: ["tried to connect to ('127.0.0.1', 38225), but an error occured: [Errno 111] Connection refused"]

根因分析 #

这个报错和内存配置不足无关，核心是Spark JVM进程与本地Python进程的Socket数据传输通道意外断连：

• 配置参数错误：spark.executor.heartbeatInterval=3600s和spark.network.timeout=3610s差值仅10s，远低于官方要求的「心跳间隔为网络超时1/10」的规范，大结果集传输时只要出现轻微网络波动就会触发连接中断，直接掐断Driver端用来接收数据的Python Socket服务。
• 原生传输机制缺陷：toPandas()默认通过单Socket通道一次性拉取全量结果到本地，当结果集数据量较大时，单通道传输时间超过Socket默认accept超时阈值就会抛出Accept timed out，后续Python进程连接本地端口自然会报Connection refused。

解决方法 #

按优先级依次尝试：

• 修正超时相关配置，把错误的心跳参数调整到合理区间，新增Python进程的超时配置：

# 替换原有配置里的超时相关项
--conf spark.executor.heartbeatInterval=60s
--conf spark.network.timeout=600s
--conf spark.worker.timeout=600s
--conf spark.python.worker.timeout=600s
--conf spark.pyspark.python.worker.reuse=true

不要把心跳间隔设为1小时这类极端值，这个参数是Driver用来检测Executor存活状态的，设置过大会导致Driver无法及时感知故障节点，反而加剧作业不稳定。

• 替换原生toPandas()调用，改为逐分区拉取合并，避免单通道长时间传输超时：

import pandas as pd

pdf_list = []
# 逐分区拉取，预取2个分区减少等待
for row_batch in SPARK_DF.rdd.toLocalIterator(prefetchPartitions=2):
    batch_df = pd.DataFrame([r.asDict(recursive=True) for r in row_batch])
    pdf_list.append(batch_df)
DF_PD = pd.concat(pdf_list, ignore_index=True)

• 如果结果集超过5G，直接绕开Socket传输链路：先把Spark DataFrame写到本地临时parquet路径，再用Pandas读取parquet文件，完全规避跨进程Socket传输的超时问题：

import pandas as pd

temp_parquet_path = "/tmp/spark_to_pandas_temp"
SPARK_DF.write.mode("overwrite").parquet(temp_parquet_path)
DF_PD = pd.read_parquet(temp_parquet_path)

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