PySpark分发模块导入

Question

在过去的几天里,我一直在努力了解Spark执行者如何知道在导入时如何使用给定名称的模块.我正在使用AWS EMR.情况:我通过键入来初始化EMR上的pyspark

pyspark - 主纱

然后,在pyspark,

import numpy as np ## notice the naming

def myfun(x):
    n = np.random.rand(1)
    return x*n

rdd = sc.parallelize([1,2,3,4], 2)
rdd.map(lambda x: myfun(x)).collect() ## works!

我的理解是,当我输入numpy as np,主节点导入和识别的唯一节点numpy通过np.然而,与EMR簇(2个工作者节点),如果我上运行RDD地图功能,驱动程序发送功能的工作节点执行用于列表中每个项目的功能(对于每个分区),和一个返回成功结果.

我的问题是:工人们如何知道numpy应该作为np导入？每个worker都已经安装了numpy,但是我没有为每个节点明确定义一种导入模块的方法as np.

有关依赖关系的更多详细信息,请参阅Cloudera的以下帖子:http: //blog.cloudera.com/blog/2015/09/how-to-prepare-your-apache-hadoop-cluster-for-pyspark-jobs/

在Complex Dependency下,他们有一个示例(代码),其中pandas模块在每个节点上显式导入.

我听说过的一个理论是驱动程序分发在pyspark交互式shell中传递的所有代码.我对此持怀疑态度.我提出的反驳这个想法的例子是,如果在主节点上我输入:

print "hello"

是每个工人节点还打印"你好"？我不这么认为.但也许我错了.

Answer 1

当函数被序列化时，会保存许多对象：

• 代码
• 全局变量
• 默认值
• 关闭
• 词典

稍后可用于恢复给定功能所需的完整环境。

由于np被函数引用，因此可以从其代码中提取：

from pyspark.cloudpickle import CloudPickler

CloudPickler.extract_code_globals(myfun.__code__)
## {'np'}

并且绑定可以从它的中提取globals：

myfun.__globals__['np']
## <module 'numpy' from ...

因此，序列化闭包（广义上）捕获了恢复环境所需的所有信息。当然，在闭包中访问的所有模块都必须可以在每台工作机器上导入。

其他一切都只是读写机器。

另外，主节点不应执行任何 Python 代码。它负责资源分配而不是运行应用程序代码。