我可以在Dask/Distributed中使用从.py文件导入的函数吗？

Question

我有关于序列化和导入的问题.

• 功能应该有自己的进口吗？就像我见过的PySpark一样
• 以下是完全错误的吗？是否mod.py需要一个畅达/ PIP封装？mod.py被写入共享文件系统.

In [1]: from distributed import Executor

In [2]: e = Executor('127.0.0.1:8786')

In [3]: e
Out[3]: <Executor: scheduler="127.0.0.1:8786" processes=2 cores=2>

In [4]: import socket

In [5]: e.run(socket.gethostname)
Out[5]: {'172.20.12.7:53405': 'n1015', '172.20.12.8:53779': 'n1016'}

In [6]: %%file mod.py
   ...: def hostname():
   ...:     return 'the hostname'
   ...: 
Overwriting mod.py

In [7]: import mod

In [8]: mod.hostname()
Out[8]: 'the hostname'

In [9]: e.run(mod.hostname)
distributed.utils - ERROR - No module named 'mod'

Answer 1

快速回答

将mod.py文件上传到所有工作人员.您可以使用您用于设置dask.distributed的任何机制来执行此操作,也可以使用upload_file方法

e.upload_file('mod.py')

或者,如果您的函数是在IPython中创建的,而不是作为模块的一部分,它将被发送而没有问题.

答案很长

这一切都与函数如何在Python中序列化有关.模块中的函数按其模块名称和函数名称进行序列化

In [1]: from math import sin

In [2]: import pickle

In [3]: pickle.dumps(sin)
Out[3]: b'\x80\x03cmath\nsin\nq\x00.'

因此,如果客户端机器想要引用math.sin它沿着这个字节串发送的函数(你会注意到它'math'并'sin'在其中掩埋在其他字节中)发送到工作机器.工作人员看着这个字节串并说"好的,我想要的功能在这样一个模块中,让我去找我的本地文件系统.如果模块不存在则会引发错误,就像你上面收到的一样.

对于动态创建的函数(在IPython中创建的函数),它使用完全不同的方法,捆绑所有代码.这种方法通常很好.

一般来说,Dask假设工作者和客户端都具有相同的软件环境.通常,这主要由使用Docker之类的其他工具设置群集的人来处理.upload_file当您拥有更频繁更新的文件或脚本时,类似的方法可以填补空白.