带未传递参数的Dask延迟函数调用

Question

我试图更好地理解使用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'",)

我是新手。我想这与序列化有关，我不了解。有人可以启发我和/或指向相关文档吗？谢谢！

Answer 1

我将尽量保持简短。

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

因此，您可以在工作器中调用read作为函数的一部分，否则，可能无法通过函数使用模式或设置模块全局变量。 Dask的延迟机制偏向于功能纯净，因此您获得的结果不应取决于运行时的当前状态。

（（如果您在主脚本中调用read之后创建了客户端，则工作器[[might拥有内存中的版本，具体取决于配置如何在系统上创建子流程）] >

Answer 2

我建议您将所有参数显式传递给迟钝的延迟函数，而不要依赖于全局名称空间。