python - PySpark 序列化 EOFError

Question

我正在读取 CSV 作为 Spark DataFrame 并对其执行机器学习操作。我不断收到 Python 序列化 EOFError - 知道为什么吗？我认为这可能是一个内存问题——即文件超出了可用 RAM——但大幅减小 DataFrame 的大小并不能防止 EOF 错误。

下面的玩具代码和错误。

#set spark context
conf = SparkConf().setMaster("local").setAppName("MyApp")
sc = SparkContext(conf = conf)
sqlContext = SQLContext(sc)

#read in 500mb csv as DataFrame
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true',
     inferschema='true').load('myfile.csv')

#get dataframe into machine learning format
r_formula = RFormula(formula = "outcome ~ .")
mldf = r_formula.fit(df).transform(df)

#fit random forest model
rf = RandomForestClassifier(numTrees = 3, maxDepth = 2)
model = rf.fit(mldf)
result = model.transform(mldf).head()

在单个节点上使用 spark-submit 重复运行上述代码会引发以下错误，即使在拟合模型之前减小了 DataFrame 的大小(例如 tinydf = df .sample(False, 0.00001):

Traceback (most recent call last):
  File "/home/hduser/spark1.6/python/lib/pyspark.zip/pyspark/daemon.py", line 157, 
     in manager
  File "/home/hduser/spark1.6/python/lib/pyspark.zip/pyspark/daemon.py", line 61, 
     in worker
  File "/home/hduser/spark1.6/python/lib/pyspark.zip/pyspark/worker.py", line 136, 
     in main if read_int(infile) == SpecialLengths.END_OF_STREAM:
  File "/home/hduser/spark1.6/python/lib/pyspark.zip/pyspark/serializers.py", line 545, 
     in read_int
    raise EOFError
  EOFError

Answer 1

该错误似乎发生在 pySpark read_int 函数中。代码如下spark site :

def read_int(stream):
length = stream.read(4)
if not length:
    raise EOFError
return struct.unpack("!i", length)[0]

这意味着当从流中读取 4 个字节时，如果读取了 0 个字节，则会引发 EOF 错误。 python 文档是 here .