Pyspark 与 Zeppelin：将文件分发到集群节点与 SparkContext.addFile()

Question

我有一个我构建的库，我希望它可供 pyspark 集群 (1.6.3) 上的所有节点使用。我通过 Zeppelin (0.7.3) 在该火花集群上运行测试程序。

我想要的文件在 github 存储库中。所以我将该存储库克隆到集群的所有节点上，并通过 pssh 创建一个脚本来同时更新它们。所以这些文件存在于每个节点上的一个设定位置，我希望每个节点都可以访问它们。

我试过了

import sys
sys.path.insert(0, "/opt/repo/folder/")

from module import function
return_rdd = function(arguments)

这产生了一个错误堆栈：

  File "/usr/hdp/current/spark-client/python/pyspark/worker.py", line 98, in main
    command = pickleSer._read_with_length(infile)
  File "/usr/hdp/current/spark-client/python/pyspark/serializers.py", line 164, in _read_with_length
    return self.loads(obj)
  File "/usr/hdp/current/spark-client/python/pyspark/serializers.py", line 439, in loads
    return pickle.loads(obj, encoding=encoding)
ImportError: No module named 'module'

我发现这个错误很不寻常，因为它是由 pickle 调用提示的。该代码似乎加载了一个数据帧并将其分区，但仅当在转换为 rdd 的分区 df 上调用模块内的另一个函数时才会失败。我不确定这里涉及泡菜电话的位置和原因；模块 pyscript 不需要被腌制，因为有问题的模块应该已经在集群的每个节点上的 sys.path 中。

另一方面，我能够做到这一点

sc.addFile("/opt/repo/folder/module.py")
import sys
from pyspark import SparkFiles
sys.path.insert(0, SparkFiles.getRootDirectory())

from module import function
return_rdd = function(arguments)

知道为什么第一种方法不起作用吗？

Answer 1

一个可能的解决方案是：

sc.addFile("/opt/repo/folder/module.py")
import sys
from pyspark import SparkFiles
sys.path.insert(0, SparkFiles.getRootDirectory())

from module import function
return_rdd = function(arguments)

这在集群模式下不起作用