为什么PySpark会随机出现"Socket is closed"错误？

Question

我刚刚参加了一个PySpark培训课程,我正在编写一个示例代码行的脚本(这解释了为什么代码块什么都不做).每次运行此代码时,我都会收到一次或两次此错误.抛出它的线在运行之间变化.我试过设置spark.executor.memory和spark.executor.heartbeatInterval,但错误依然存在.我也尝试过.cache()各种各样的行,不做任何修改.

错误:

16/09/21 10:29:32 ERROR Utils: Uncaught exception in thread stdout writer for python
java.net.SocketException: Socket is closed
        at java.net.Socket.shutdownOutput(Socket.java:1551)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply$mcV$sp(PythonRDD.scala:344)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply(PythonRDD.scala:344)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply(PythonRDD.scala:344)
        at org.apache.spark.util.Utils$.tryLog(Utils.scala:1870)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3.apply(PythonRDD.scala:344)
        at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1857)
        at org.apache.spark.api.python.PythonRunner$WriterThread.run(PythonRDD.scala:269)

代码:

from pyspark import SparkConf, SparkContext

def parseLine(line):
    fields = line.split(',')
    return (int(fields[0]), float(fields[2]))

def parseGraphs(line):
    fields = line.split()
    return (fields[0]), [int(n) for n in fields[1:]]

# putting the [*] after local makes it run one executor on each core of your local PC
conf = SparkConf().setMaster("local[*]").setAppName("MyProcessName")

sc = SparkContext(conf = conf)

# parse the raw data and map it to an rdd.
# each item in this rdd is a tuple
# two methods to get the exact same data:
########## All of these methods can use lambda or full methods in the same way ##########
# read in a text file
customerOrdersLines = sc.textFile("file:///SparkCourse/customer-orders.csv")
customerOrdersRdd = customerOrdersLines.map(parseLine)
customerOrdersRdd = customerOrdersLines.map(lambda l: (int(l.split(',')[0]), float(l.split(',')[2])))
print customerOrdersRdd.take(1)

# countByValue groups identical values and counts them
salesByCustomer = customerOrdersRdd.map(lambda sale: sale[0]).countByValue()
print salesByCustomer.items()[0]

# use flatMap to cut everything up by whitespace
bookText = sc.textFile("file:///SparkCourse/Book.txt")
bookRdd = bookText.flatMap(lambda l: l.split())
print bookRdd.take(1)

# create key/value pairs that will allow for more complex uses
names = sc.textFile("file:///SparkCourse/marvel-names.txt")
namesRdd = names.map(lambda line: (int(line.split('\"')[0]), line.split('\"')[1].encode("utf8")))
print namesRdd.take(1)

graphs = sc.textFile("file:///SparkCourse/marvel-graph.txt")
graphsRdd = graphs.map(parseGraphs)
print graphsRdd.take(1)

# this will append "extra text" to each name.
# this is faster than a normal map because it doesn't give you access to the keys
extendedNamesRdd = namesRdd.mapValues(lambda heroName: heroName + "extra text")
print extendedNamesRdd.take(1)

# not the best example because the costars is already a list of integers
# but this should return a list, which will update the values
flattenedCostarsRdd = graphsRdd.flatMapValues(lambda costars: costars)
print flattenedCostarsRdd.take(1)

# put the heroes in ascending index order
sortedHeroes = namesRdd.sortByKey()
print sortedHeroes.take(1)

# to sort heroes by alphabetical order, we switch key/value to value/key, then sort
alphabeticalHeroes = namesRdd.map(lambda (key, value): (value, key)).sortByKey()
print alphabeticalHeroes.take(1)

# make sure that "spider" is in the name of the hero
spiderNames = namesRdd.filter(lambda (id, name): "spider" in name.lower())
print spiderNames.take(1)

# reduce by key keeps the key and performs aggregation methods on the values.  in this example, taking the sum
combinedGraphsRdd = flattenedCostarsRdd.reduceByKey(lambda value1, value2: value1 + value2)
print combinedGraphsRdd.take(1)

# broadcast: this is accessible from any executor
sentData = sc.broadcast(["this can be accessed by all executors", "access it using sentData"])

# accumulator:  this is synced across all executors
hitCounter = sc.accumulator(0)

Answer 1

免责声明：我还没有在 Spark 代码库的这一部分上花费足够的时间，但是让我给您一些可能会导致解决方案的提示。以下内容只是解释在哪里搜索更多信息，而不是问题的解决方案。

您面临的异常是由于此处代码中所示的其他问题造成的（正如您可能在执行java.net.Socket.shutdownOutput(Socket.java:1551)时看到的行）。worker.shutdownOutput()

16/09/21 10:29:32 ERROR Utils: Uncaught exception in thread stdout writer for python
java.net.SocketException: Socket is closed
        at java.net.Socket.shutdownOutput(Socket.java:1551)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply$mcV$sp(PythonRDD.scala:344)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply(PythonRDD.scala:344)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3$$anonfun$apply$4.apply(PythonRDD.scala:344)
        at org.apache.spark.util.Utils$.tryLog(Utils.scala:1870)
        at org.apache.spark.api.python.PythonRunner$WriterThread$$anonfun$run$3.apply(PythonRDD.scala:344)
        at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1857)
        at org.apache.spark.api.python.PythonRunner$WriterThread.run(PythonRDD.scala:269)

这使我相信该错误是其他一些早期错误的后续错误。

python 的 stdout writer名称是负责 Spark 和 pyspark 之间通信的线程的名称（使用EvalPythonExec物理运算符）（因此您无需进行太多更改即可执行 python 代码）。

事实上，scaladocEvalPythonExec提供了有关 pyspark 内部使用的底层通信基础设施以及使用套接字连接到外部 Python 进程的大量信息。

Python 评估的工作原理是通过套接字将必要的（投影的）输入数据发送到外部 Python 进程，并将 Python 进程的结果与原始行相结合。

此外，python默认情况下使用除非使用PYSPARK_DRIVER_PYTHONor覆盖（如您在shell 脚本中PYSPARK_PYTHON看到的那样）。这是出现在失败线程名称中的名称。pyspark

16/09/21 10:29:32 错误实用程序： Python的线程标准输出编写器中未捕获异常

我建议使用以下命令检查系统上的 python 版本。

python -c 'import sys; print(sys.version_info)'

这应该是 Python 2.7+，但可能您使用的是未经 Spark 良好测试的最新 Python。猜测...

您应该包含 pyspark 应用程序执行的整个日志，这就是我希望找到答案的地方。