ModuleNotFoundError 因为 PySpark 序列化程序无法找到库文件夹

Posted 2023-02-16

【中文标题】ModuleNotFoundError 因为 PySpark 序列化程序无法找到库文件夹

【英文标题】：ModuleNotFoundError because PySpark serializer is not able to locate library folder

【发布时间】：2019-05-20 16:10:45

【问题描述】：

我有以下文件夹结构

 - libfolder
    - lib1.py
    - lib2.py
 - main.py

main.py 调用 libfolder.lib1.py 然后调用 libfolder.lib2.py 和其他人。

在本地机器上一切正常，但在我将它部署到 Dataproc 后，我收到以下错误

File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 455, in loads
return pickle.loads(obj, encoding=encoding)
ModuleNotFoundError: No module named 'libfolder'

我已将文件夹压缩到xyz.zip 并运行以下命令：

spark-submit --py-files=xyz.zip main.py

序列化程序无法找到 libfolder 的位置。我打包文件夹的方式有问题吗？

此问题与this one 类似，但未得到解答。

编辑：回复伊戈尔的问题

unzip -l 为 zip 文件返回以下内容

 - libfolder
    - lib1.py
    - lib2.py
 - main.py

在 main.py 中，lib1.py 是用这个 import 语句调用的

from libfolder import lib1

【问题讨论】：

能否列出 zip 文件 (unzip -l xyz.zip) 的内容，并说明 main.py 在代码中如何调用 libfolder.lib1.py？

另外请说明您如何调用 dataproc？

你是从哪个文件夹运行这个命令的？

【参考方案1】：

这对我有用：

$ cat main.py

from pyspark import SparkContext, SparkConf

from subpkg import sub

conf = SparkConf().setAppName("Shell Count")
sc = SparkContext(conf = conf)

text_file = sc.textFile("file:///etc/passwd")
counts = text_file.map(lambda line: sub.map(line)) \
    .map(lambda shell: (shell, 1)) \
    .reduceByKey(lambda a, b: sub.reduce(a, b))

counts.saveAsTextFile("hdfs:///count5.txt")

$ cat subpkg/sub.py

def map(line):
  return line.split(":")[6]

def reduce(a, b):
  return a + b

$ unzip -l /tmp/deps.zip 
Archive:  /tmp/deps.zip
  Length      Date    Time    Name
---------  ---------- -----   ----
        0  2019-01-07 14:22   subpkg/
        0  2019-01-07 13:51   subpkg/__init__.py
       79  2019-01-07 14:13   subpkg/sub.py
---------                     -------
       79                     3 files


$ gcloud dataproc jobs submit pyspark --cluster test-cluster main.py --py-files deps.zip
Job [1f0f15108a4149c5942f49513ce04440] submitted.
Waiting for job output...
Hello world!
Job [1f0f15108a4149c5942f49513ce04440] finished successfully.

【讨论】：

它适用于 noop 脚本。只有使用 spark 的 reduce 方法时才会出现上述错误

嗯，如果我将 map/reduce 函数提取到另一个模块中，那么简单的字数统计程序也适用于我。可以分享一下sn-p代码吗？

【参考方案2】：

对我来说，在引用中写 py-files。

spark2-submit --master yarn --verbose --deploy-mode client \
  --py-files "dependencies.zip" $HOME/jobs/master/etl_master.py

还要记住参数的顺序，我建议先添加--verbose，然后再次提交作业并查看日志以找到您的程序传递给服务器的参数。