現(xiàn)象

如果我們安裝了python3.7版本來運行spark，并且用到了RDD的take函數(shù)，就會報錯：RuntimeError:?generator?raised?StopIteration

我們可以編寫一個python腳本test.py來進(jìn)行測試，用spark-submit命令提交： spark-submit test.pyfrom?pyspark.sql?import?SparkSession

spark?=?SparkSession\

.builder\

.appName("PythonWordCount")\

.getOrCreate()

sc?=?spark.sparkContext

print(sc.parallelize([1,?2]).take(1))

當(dāng)然也可以直接在pyspark的交互式shell中執(zhí)行sc.parallelize([1,?2]).take(1)

執(zhí)行就會碰到上面說的錯誤。

原因分析

這個錯誤是由于Python3.7合入了一個不兼容性修改PEP-0479引起的。

Spark社區(qū)已經(jīng)修復(fù)并合入了Spark的2.3.2和2.4版本，具體參考Spark對應(yīng)的JIRA?SPARK-24739。

解決方法

MRS目前使用的版本為Spark 2.2.1，尚未修復(fù)。目前可以通過兩種方法規(guī)避：（推薦）使用Python 3.6，3.5；

參考開源合入，修改提交任務(wù)節(jié)點的對應(yīng)python腳本，并打包。具體見下一節(jié)。

參考開源合入，修改不兼容的python代碼

打開客戶端節(jié)點，修改/opt/client/Spark/spark/python/pyspark/rdd.py中的takeUpToNumLeft方法。def?takeUpToNumLeft(iterator):

iterator?=?iter(iterator)

taken?=?0

while?taken?<?left:

try:

yield?next(iterator)

except?StopIteration:

return

taken?+=?1

重新打包pyspark.zip#?備份pyspark.zip

cd?/opt/client/Spark/spark/python/lib?&&?mv?pyspark.zip?pyspark.zip.bak

#?打包新的pyspark.zip

cd?/opt/client/Spark/spark/python;?zip?-r?pyspark.zip?pyspark

#?移動至lib目錄

mv?pyspark.zip?/opt/client/Spark/spark/python/lib/

#?修改權(quán)限

chmod?777?/opt/client/Spark/spark/python/lib/pyspark.zip

這時再來試一下。spark-submit?test.py

或者

spark-submit?--master?yarn?test.py

額外說一點，使用pyspark

shell，使用的是未打包的python，就是/opt/client/Spark/spark/python/pyspark/xxx.py；使用spark-submit命令提交，用的就是lib目錄下的pyspark.zip了，具體可以看下面這一行上傳的日志。2018-12-18?16:24:05,888?|?INFO??|?Thread-3?|?Uploading?resource?file:/opt/client/Spark/spark/python/li

b/pyspark.zip?->?hdfs://hacluster/user/hdfs/.sparkStaging/application_1545119010002_0007/pyspark.zip?|

org.apache.spark.internal.Logging$class.logInfo(Logging.scala:54)

總結(jié)

以上是生活随笔為你收集整理的spark调用python_在MRS集群中使用Python3.7运行PySpark程序，调用RDD的take函数报错处理...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。