ModuleNotFoundError, поскольку сериализатор PySpark не может найти папку библиотеки

Question

У меня следующая структура папок

 - 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, Есть ли проблема с тем, как я упаковываю свои папки?

Эта проблема похожа на эту, но не получила ответа.

Изменить: ответ на вопросы Игоря

unzip -l для zip-файла возвращает следующее

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

В main.py lib1.py вызывается с этим оператором импорта

from libfolder import lib1

Answer 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.

Answer 2

Вам не нужно передавать zip-файл, используя параметр --py-files в spark-submit . Вам просто нужно импортировать пакет, в котором функция UDF находится в вашем проекте Python. Например:

      from package1.subpackage1.UDFPythonFile1 import UDF1

Package1 — это каталог, в котором вы выполняете скрипт Python точки входа (main.py)

• main.py
• пакет1
  • подпакет1
    • UDFPythonFile1.py

Преимущество реализации вышеуказанного подхода заключается в том, чтобы избежать создания zip-файла каждый раз. Это сработало для меня на машине с Windows.

Answer 3

Для меня работает написание py-файлов в цитате.

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

Также запомните порядок аргументов, я бы рекомендовал сначала добавить --verbose после этого снова отправьте задание и просмотрите журнал, чтобы найти аргументы, передаваемые вашей программой на сервер.