在dask Client.map调用期间会发生什么？

如何解决在dask Client.map调用期间会发生什么？

我正在尝试使用dask编写网格搜索实用程序。目标函数调用包含大量数据源的类的方法。我试图使用dask将计算并行化为多核解决方案，而不必复制原始的类/数据帧。我在文档中没有找到任何解决方案，因此我在这里发布了一个玩具示例：

import pickle
from dask.distributed import Client,LocalCluster
from multiprocessing import current_process


class TestClass:
    def __init__(self):
        self.param = 0

    def __getstate__(self):
        print("I am pickled!")
        return self.__dict__

    def loss(self,ext_param):
        self.param += 1
        print(f"{current_process().pid}: {hex(id(self))}:  {self.param}: {ext_param} ")
        return f"{self.param}_{ext_param}"


def objective_function(param):
    return test_instance.loss(param)

if __name__ == '__main__':

    test_instance = TestClass()
    print(hex(id(test_instance)))
    cluster = LocalCluster(n_workers=2)
    client = Client(cluster)
    futures = client.map(objective_function,range(20))
    result = client.gather(futures)
    print(result)
    
# ---- OUTPUT RESULTS ----
# 0x7fe0a5056d30
# I am pickled!
# I am pickled!
# 11347: 0x7fb9bcfa0588:  1: 0
# 11348: 0x7fb9bd0a2588:  1: 1
# 11347: 0x7fb9bcf94240:  1: 2
# 11348: 0x7fb9bd07b6a0:  1: 3
# 11347: 0x7fb9bcf945f8:  1: 4 
# ['1_0','1_1','1_2','1_3','1_4']

我有以下问题：

1. 为什么下面的pickle函数被调用两次？
2. 我注意到map函数的每个迭代都使用test_instance的新副本，如您从每个迭代的不同类地址以及{{1 }}属性在每次迭代时都设置为0（此行为与多处理的标准实现不同。我突出显示了here池）。我假设在每次迭代过程中，每个进程都会收到一个新的腌制类副本-是正确的吗？
3. 从（2）开始，计算过程中内存中有test_instance.param个副本？是1（对于主线程中的原始实例）+1（经过酸洗的副本）+2（每个进程中存在的实例）= 4？有什么办法可以将此值设为1？

我注意到this github issue中提出的使用Ray库可以使用某些共享内存解决方案的方法。

解决方法

为什么下面的pickle函数被调用两次？

通常，python的pickle有效地将实例变量和引用绑定到导入模块中的类。在__main__中，这可能是不可靠的，并且dask退回到了cloudpickle（内部也称为pickle）。在我看来，在第一次尝试腌制之前，可能会发生在"__main__"中进入distributed.protocol.pickle.dumps的检查。

在每次迭代过程中，每个进程将收到一个新的腌制类副本

是的。每次dask运行任务时，它都会反序列化输入，从而创建实例的nw副本。请注意，您的熟练工可能是通过fork_server技术创建的，因此内存不是简单地复制的（这是做事的安全方法）。

您可以在计算之前将实例“分散”给工作人员，他们可以重用其本地副本，但不应通过更改对象而是返回结果（即从功能上）来执行轻松的任务。

内存中有多少test_instance副本

客户端中的

1个，每个正在执行的任务加一个。序列化的版本也可能存在，可能是在图中保留的版本，暂时保留在客户端上，然后保留在调度程序上。反序列化时，它也将临时存储在工作人员内存中。对于某些类型，可以使用零拷贝de / ser。

如果由于对象的大小而导致任务非常大，则一定要事先“分散”它们（client.scatter）。

有什么办法可以将该值设为1？

您可以在进程中运行调度程序和/或工作程序以共享内存，但是，当然，您会失去与GIL的并行性。

也许您可以尝试Actor interface？该模式似乎与您的工作流程相符。