相关疑难解决方法(0)

我在Spark上使用Python时遇到了问题.我的应用程序有一些依赖项,例如numpy,pandas,astropy等.我不能使用virtualenv来创建一个包含所有依赖项的环境,因为除了HDFS之外,集群上的节点没有任何公共mountpoint或文件系统.因此我坚持使用spark-submit --py-files.我将site-packages的内容打包到一个ZIP文件中并像使用--py-files=dependencies.zip选项一样提交作业(最简单的方法是在Spark执行器节点上安装Python依赖项？).但是,群集上的节点似乎仍然没有看到内部的模块,并且ImportError在导入numpy时它们会抛出这样的内容.

File "/path/anonymized/module.py", line 6, in <module>
    import numpy
File "/tmp/pip-build-4fjFLQ/numpy/numpy/__init__.py", line 180, in <module>   
File "/tmp/pip-build-4fjFLQ/numpy/numpy/add_newdocs.py", line 13, in <module>
File "/tmp/pip-build-4fjFLQ/numpy/numpy/lib/__init__.py", line 8, in <module>
    #
File "/tmp/pip-build-4fjFLQ/numpy/numpy/lib/type_check.py", line 11, in <module>
File "/tmp/pip-build-4fjFLQ/numpy/numpy/core/__init__.py", line 14, in <module>
ImportError: cannot import name multiarray

当我切换到virtualenv并使用本地pyspark shell时,一切正常,所以依赖都在那里.有谁知道,什么可能导致这个问题,以及如何解决它？

谢谢!

我的目标是将自定义.py文件导入我的spark应用程序并调用该文件中包含的一些函数

这是我尝试过的:

我有一个名为Test.py的测试文件,如下所示:

def func():
    print "Import is working"

在我的Spark应用程序中,我执行以下操作(如文档中所述):

sc = SparkContext(conf=conf, pyFiles=['/[AbsolutePathTo]/Test.py'])

我也尝试了这个(在创建Spark上下文之后):

sc.addFile("/[AbsolutePathTo]/Test.py")

在提交我的spark应用程序时,我甚至尝试过以下方法:

./bin/spark-submit --packages com.datastax.spark:spark-cassandra-connector_2.10:1.5.0-M2 --py-files /[AbsolutePath]/Test.py ../Main/Code/app.py

但是,我总是得到一个名称错误:

NameError: name 'func' is not defined

当我在app.py中调用func()时.(如果我尝试调用Test.func(),则与'Test'相同的错误)

最后,我还尝试使用与上面相同的命令在pyspark shell中导入文件:

sc.addFile("/[AbsolutePathTo]/Test.py")

奇怪的是,我没有在导入时出错,但仍然,我不能在没有得到错误的情况下调用func().此外,不确定它是否重要,但我在一台机器上本地使用火花.

我真的尝试了我能想到的一切,但仍然无法让它发挥作用.可能我错过了一些非常简单的事情.任何帮助,将不胜感激.

在我的应用程序的代码库中集成pyspark时,我无法在RDD的map方法中引用类的方法.我用一个简单的例子重复了这个问题,如下所示

这是一个虚拟类,我已经定义了它只是为RDD的每个元素添加一个数字,RDD是一个类属性:

class Test:

    def __init__(self):
        self.sc = SparkContext()
        a = [('a', 1), ('b', 2), ('c', 3)]
        self.a_r = self.sc.parallelize(a)

    def add(self, a, b):
        return a + b

    def test_func(self, b):
        c_r = self.a_r.map(lambda l: (l[0], l[1] * 2))
        v = c_r.map(lambda l: self.add(l[1], b))
        v_c = v.collect()
        return v_c

test_func()map()在RDD上调用方法v,然后在add()每个元素上调用方法v.调用test_func()抛出以下错误:

pickle.PicklingError: Could not serialize object: Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, …