使用未传递参数的 Dask 延迟函数调用答案

【问题标题】：Dask delayed function call with non-passed parameters使用未传递参数的 Dask 延迟函数调用
【发布时间】：2020-07-14 22:11:23
【问题描述】：

我正在寻求更好地理解使用dask.delayed 调用依赖于参数的函数时的以下行为。当在 configparser 读取的参数文件中指定参数时，似乎会出现此问题。这是一个完整的例子：

参数文件：

#zpar.ini: parameter file for configparser

[my pars]
my_zpar = 2.

解析器：

#zippy_parser
import configparser

def read(_rundir):

    global rundir
    rundir = _rundir

    cp = configparser.ConfigParser()
    cp.read(rundir + '/zpar.ini')

    #[my pars]
    global my_zpar
    my_zpar = cp['my pars'].getfloat('my_zpar')

和主要的python文件：

# dask test with configparser
import dask
from dask.distributed import Client
import zippy_parser as zpar


def my_func(x, y):

    # print stuff
    print("parameter from main is: {}".format(main_par))
    print("parameter from configparser is: {}".format(zpar.my_zpar))

    # do stuff
    return x + y


if __name__ == '__main__':

    client = Client(n_workers = 4)

    #read parameters from input file
    rundir = '/path/to/parameter/file'
    zpar.read(rundir)

    #test zpar
    print("zpar is {}".format(zpar.my_zpar))

    #define parameter and call my_func
    main_par = 5.
    z = dask.delayed(my_func)(1., 2.)
    z.compute()

    client.close()

my_func() 中的第一个 print 语句执行得很好，但是第二个 print 语句引发了异常。输出为：

zpar is 2.0
parameter from main is: 5.0
distributed.worker - WARNING -  Compute Failed
Function:  my_func
args:      (1.0, 2.0)
kwargs:    {}
Exception: AttributeError("module 'zippy_parser' has no attribute 'my_zpar'",)

我是新手。我想这与序列化有关，我不明白。有人可以启发我和/或指出相关文件吗？谢谢！

【问题讨论】：

你的例子很复杂（很多文件，很多重定向），但不完整：|比如zippy.py没有定义_zippy_sum_array，不能按写的导入zippy_parser或zstuff，除非你在正确的目录下；并且永远不会调用 read。
天哪。我知道。我试图在不使帖子更长的情况下包含我认为是必需品的内容。我会尽量精简我的例子。
@mdurant 感谢您的反馈。我编辑了原始帖子。应该是完整的、更简单的，并且仍然可以解决主要问题。

标签： python dask configparser dask-delayed

【解决方案1】：

我会尽量保持简短。

当一个函数被序列化以便发送给工作人员时，python 也会发送函数所需的局部变量和函数（它的“闭包”）。但是，它按名称存储它引用的模块，它不会尝试序列化您的整个运行时。这意味着 zippy_parser 在工作线程中导入，而不是反序列化。由于函数read 从未被调用过在 worker 中，global 变量永远不会被初始化。

因此，您可以在工作程序中调用 read 作为函数的一部分或以其他方式调用，但使用函数的模式或设置模块全局变量可能不是很好。 Dask 的延迟机制更喜欢功能纯度，即您获得的结果不应依赖于运行时的当前状态。

（请注意，如果您在主脚本中调用 read 后创建了客户端，则工作人员可能拥有内存版本，具体取决于子进程的配置方式你的系统）

【讨论】：

【解决方案2】：

我鼓励您将所有参数显式传递给您的 dask 延迟函数，而不是依赖全局命名空间。

【讨论】：