Pyspark --py-files 不起作用答案

【问题标题】：Pyspark --py-files doesn't workPyspark --py-files 不起作用
【发布时间】：2017-05-22 13:53:23
【问题描述】：

我使用这个作为文件建议http://spark.apache.org/docs/1.1.1/submitting-applications.html

spsark 版本 1.1.0

./spark/bin/spark-submit --py-files /home/hadoop/loganalysis/parser-src.zip \
/home/hadoop/loganalysis/ship-test.py

和代码中的conf：

conf = (SparkConf()
        .setMaster("yarn-client")
        .setAppName("LogAnalysis")
        .set("spark.executor.memory", "1g")
        .set("spark.executor.cores", "4")
        .set("spark.executor.num", "2")
        .set("spark.driver.memory", "4g")
        .set("spark.kryoserializer.buffer.mb", "128"))

和从节点抱怨 ImportError

14/12/25 05:09:53 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, ip-172-31-10-8.cn-north-1.compute.internal): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/home/hadoop/spark/python/pyspark/worker.py", line 75, in main
    command = pickleSer._read_with_length(infile)
  File "/home/hadoop/spark/python/pyspark/serializers.py", line 150, in _read_with_length
    return self.loads(obj)
ImportError: No module named parser

并且 parser-src.zip 在本地测试。

[hadoop@ip-172-31-10-231 ~]$ python
Python 2.7.8 (default, Nov  3 2014, 10:17:30) 
[GCC 4.8.2 20140120 (Red Hat 4.8.2-16)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import sys
>>> sys.path.insert(1, '/home/hadoop/loganalysis/parser-src.zip')
>>> from parser import parser
>>> parser.parse
<function parse at 0x7fa5ef4c9848>
>>>

我正在尝试获取有关远程工作人员的信息。看看它是否复制了文件。sys.path 是什么样子的。这很棘手。

更新：我用这个发现 zip 文件已发货。并设置了 sys.path。仍然导入得到错误。

data = list(range(4))
disdata = sc.parallelize(data)
result = disdata.map(lambda x: "sys.path:  {0}\nDIR: {1}   \n FILES: {2} \n parser: {3}".format(sys.path, os.getcwd(), os.listdir('.'), str(parser)))
result.collect()
print(result.take(4))

看来我必须深入研究 cloudpickle。这意味着我需要先了解 cloudpickle 如何工作和失败。

: An error occurred while calling o40.collect.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 0.0 failed 4 times, most recent failure: Lost task 4.3 in stage 0.0 (TID 23, ip-172-31-10-8.cn-north-1.compute.internal): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/home/hadoop/spark/python/pyspark/worker.py", line 75, in main
    command = pickleSer._read_with_length(infile)
  File "/home/hadoop/spark/python/pyspark/serializers.py", line 150, in _read_with_length
    return self.loads(obj)
  File "/home/hadoop/spark/python/pyspark/cloudpickle.py", line 811, in subimport
    __import__(name)
ImportError: ('No module named parser', <function subimport at 0x7f219ffad7d0>, ('parser.parser',))

更新：

有人在 spark 0.8 中遇到同样的问题 http://apache-spark-user-list.1001560.n3.nabble.com/pyspark-Importing-other-py-files-in-PYTHONPATH-td2301.html

但他将他的库放在 python dist-packages 和导入工程中。我试过了，但仍然出现导入错误。

更新：

OH.gush .. 我认为问题是由于不理解 zip 文件和 python 导入行为引起的。我将 parser.py 传递给 --py-files，它可以工作，抱怨另一个依赖项。并且只压缩 .py 文件[不包括 .pyc] 似乎也可以。

但我不太明白为什么。

【问题讨论】：

标签： python hadoop apache-spark emr

【解决方案1】：

试试SparkContext这个功能

sc.addPyFile(path)

根据pyspark文档here

为将来要在此 SparkContext 上执行的所有任务添加 .py 或 .zip 依赖项。传递的路径可以是本地文件、HDFS（或其他 Hadoop 支持的文件系统）中的文件，也可以是 HTTP、HTTPS 或 FTP URI。

尝试将您的 python 模块文件上传到公共云存储（例如 AWS S3）并将 URL 传递给该方法。

这里有更全面的阅读材料：http://www.cloudera.com/documentation/enterprise/5-5-x/topics/spark_python.html

【讨论】：

正是我想要的！谢谢！

【解决方案2】：

尝试从方法本身内部而不是在驱动程序脚本的顶部导入您的自定义模块，例如：

def parse_record(record):
    import parser
    p = parser.parse(record)
    return p

而不是

import parser
def parse_record(record):
    p = parser.parse(record)
    return p

Cloud Pickle 似乎无法识别何时导入了自定义模块，因此它似乎尝试将顶级模块与运行该方法所需的其他数据一起腌制。根据我的经验，这意味着顶级模块似乎存在，但它们缺少可用的成员，并且嵌套模块无法按预期使用。一旦使用 from A import * 或从方法内部 (import A.B) 导入，模块就会按预期工作。

【讨论】：

【解决方案3】：

听起来一个或多个节点配置不正确。集群上的所有节点是否都具有相同的 Python 版本/配置（即它们都安装了解析器模块）？

如果您不想逐一检查，您可以编写一个脚本来检查它是否已为您安装/安装。 This 线程显示了一些方法来做到这一点。

【讨论】：

我使用 aws emr，并且所有节点都有引导操作脚本。它们都具有完全相同的配置/版本。都有 Python 2.7.8。

【解决方案4】：

您需要使用 setuptools 等工具打包 Python 代码。这将让您创建一个类似于 java jar 文件的 .egg 文件。然后你可以使用 --py-files 指定这个 egg 文件的路径

spark-submit --py-files path_to_egg_file path_to_spark_driver_file

【讨论】：

【解决方案5】：

我遇到了类似的问题，即使我使用--py-files 开关，我的工作节点也无法检测到模块。

我做了几件事 - 首先我尝试在创建 SparkContext (sc) 变量后放置导入语句，希望导入应该在模块发送到所有节点后进行，但它仍然不起作用。然后我尝试sc.addFile 在脚本本身中添加模块（而不是将其作为命令行参数发送），然后导入模块的功能。至少在我的情况下，这起到了作用。

【讨论】：

你是如何从你的主要 python 脚本中引用这些的？
@CarlosBribiescas 您是说我为模块使用了哪个路径？我刚刚在 sc.addFile(pathname) 中提到了我的 module.py 文件的绝对路径
这在集群上不起作用，因为 --py-files 将它们上传到服务器上的不同目录。我最终通过使用 execfile("./__pyfiles__/module.py") 来引用它们，我猜这是 spark-submit 放置文件的地方。
@CarlosBribiescas 当我指的是模块的路径时。我不检查 py-files 在服务器上上传它们的路径。我根本不使用 py-files 开关。我刚刚使用了 sc.addFile(filepath) 其中 filepath 是我创建并保存在本地磁盘某处的模块的路径。

【解决方案6】：

默认情况下，EMR 上的 PySpark 是为 Python 2.6 配置的，因此请确保没有为 Python 2.7 解释器安装它们

【讨论】：

【解决方案7】：

创建包含所有依赖项的 zip 文件 (example-abc.zip)。

在创建 spark 上下文时，将 zip 文件名提及为：

    sc = SparkContext(conf=conf, pyFiles=["abc.zip"])

【讨论】：