在 dask 分布式工作器上初始化状态

Question

我正在尝试做类似的事情

resource = MyResource()

def fn(x):

   something = dosemthing(x, resource)

   return something

client = Client()

results = client.map(fn, data)

问题是它resource不可序列化并且构建起来很昂贵。因此，我想在每个工人上构造一次它，并可供fn.

我该怎么做呢？或者有没有其他方法可以让resource所有工人都可以使用？

Answer 1

你总是可以构建一个懒惰的资源，比如class GiveAResource():    resource = [None]    def get_resource(self):        if self.resource[0] is None:            self.resource[0] = MyResource()        return self.resource[0]这样的一个实例将在进程之间很好地序列化，因此您可以将它作为要在工作人员上执行的任何函数的输入，然后调用.get_resource()它会获得本地昂贵的资源（它将在稍后出现的任何工作人员上重新制作）。此类最好在模块中而不是动态代码中定义。这里没有锁定，因此如果多个线程在到目前为止还不需要资源的情况下同时请求资源，您将获得多余的工作。