出现错误：原因：java.net.SocketTimeoutException：接受超时

Posted 2023-04-15

【中文标题】出现错误：原因：java.net.SocketTimeoutException：接受超时

【英文标题】：Getting error : Caused by: java.net.SocketTimeoutException: Accept timed out

【发布时间】：2019-03-27 03:59:59

【问题描述】：

使用以下代码在 Jupyter Notebook 中使用 python 3.7 运行 pyspark 时出现错误。

from pyspark import SparkContext
from pyspark.sql.session import SparkSession
from pyspark.sql import SQLContext
import pyspark as ps

conf = ps.SparkConf().setMaster("yarn-client").setAppName("sparK-mer")
conf.set("spark.executor.heartbeatInterval","3600s")
sc = SparkContext('local') 
sqlContext = SQLContext(sc)
from pyspark.mllib.linalg import Vector, Vectors
from nltk.stem.wordnet import WordNetLemmatizer
from pyspark.ml.feature import RegexTokenizer, StopWordsRemover, Word2Vec

我正在读取基于以下代码的csv文件：

datanew = sqlContext.read.format("csv") \
   .options(header='true', inferschema='true') \
   .load("C://Users//mypath//data.csv")

parts = datanew.rdd.map(lambda l: l.split(","))
datapysp = parts.map(lambda p: Row(uiid=p[0],title=(p[3].strip()),text=(p[4].strip())))
schemaString = "uiid title text"
fields = [StructField(field_name, StringType(), True) for  field_name in schemaString.split()]
schema = StructType(fields)
sqlContext.createDataFrame(datapysp, schema).show()

这是错误消息，我收到了，其中提到了 UIID、标题和文本的列。

Py4JJavaError: An error occurred while calling o74.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 1 times, most recent failure: Lost task 0.0 in stage 2.0 (TID 2, localhost, executor driver): org.apache.spark.SparkException: Python worker failed to connect back.
    at org.apache.spark.api.python.PythonWorkerFactory.createSimpleWorker(PythonWorkerFactory.scala:170)
    at org.apache.spark.api.python.PythonWorkerFactory.create(PythonWorkerFactory.scala:97)
    at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:117)
    at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:108)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:65)
    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.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.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.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:402)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: java.net.SocketTimeoutException: Accept timed out
    at java.net.DualStackPlainSocketImpl.waitForNewConnection(Native Method)
    at java.net.DualStackPlainSocketImpl.socketAccept(Unknown Source)
    at java.net.AbstractPlainSocketImpl.accept(Unknown Source)
    at java.net.PlainSocketImpl.accept(Unknown Source)
    at java.net.ServerSocket.implAccept(Unknown Source)
    at java.net.ServerSocket.accept(Unknown Source)
    at org.apache.spark.api.python.PythonWorkerFactory.createSimpleWorker(PythonWorkerFactory.scala:164)

我浏览了此处提供的答案： Pyspark socket timeout exception after application running for a while。我尝试根据提供的答案将代码更改为此。

import pyspark as ps

conf = ps.SparkConf().setMaster("yarn-client").setAppName("sparK-mer")
conf.set("spark.executor.heartbeatInterval","3600s")
sc = ps.SparkContext('local[4]', '', conf=conf)

在运行此部分 sc = ps.SparkContext('local[4]', '', conf=conf) 时，我收到错误消息，指出 Java 网关进程在发送其端口号之前已退出。

也尝试过这样，但我收到的关于接受超时的错误仍然相同。

 parts = datanew.rdd.map(lambda l: l.split(","))
    datapysp = parts.map(lambda p: Row(uiid=p[0],title=(p[3].strip()),text=(p[4].strip())))
    schemaString = "uiid title text"
    fields = [StructField(field_name, StringType(), True) for  field_name in schemaString.split()]
    schema = StructType(fields)
    sqlContext.createDataFrame(datapysp, 
    schema).show().config("sqlContext.executor.heartbeatInterval", "10000s") 
    --added this but still the error is not being resolved. 

如果有人可以帮助我，我将不胜感激。我使用的是 Windows 10 64 位。

【问题讨论】：

可以发布一些可能对您有所帮助的链接。没事吧？

【参考方案1】：

根据this website：

spark.executor.heartbeatInterval 10s 每个执行器对驱动程序的心跳之间的间隔。 Heartbeats 让驱动程序知道执行程序仍然存在，并使用正在进行的任务的指标对其进行更新。

看到这一点，我相信您的代码的 spark.executor.heartbeatInterval 部分存在问题。我建议你增加spark.executor.heartbeatInterval。

【讨论】：

感谢@Xilpex，我将它增加到 36000s，它仍然是一样的。 conf = ps.SparkConf().setMaster("yarn-client").setAppName("sparK-mer") newconf=conf.set("spark.executor.heartbeatInterval","36000s") sc = SparkContext('local' ,newconf) 但它仍然是同样的错误

【参考方案2】：

首先用这个 sn-p 代码检查你的 spark 版本：

print( "spark version=" ,SparkSession.builder.appName("test").getOrCreate().version)

和带有 pip 命令行的 pyspark 版本：

pip show pyspark

如果 pyspark 的版本高于你的 spark 版本， 然后你可以通过使用这个 sn-p 来使用旧版本的 pyspark：

import pkg_resources
pkg_resources.require("pyspark==[older_version]")

或者在你的系统中安装新版本的 spark