Dask似乎在为全局变量共享内存，我认为这是不可能的

Question

这里有一些可复制的代码：

1. 这是带有全局变量的Dask：

# method_file.py
import dask
import random
class TestClass():
def from_dataframe(self, pdf):
global data
data = pdf
@staticmethod
def work(elem):
data.iloc[0, 0] = 9999
return len(data) + elem
def do(self):
tasks = [dask.delayed(TestClass.work)(random.randint(1,500)) for x in range(10)]
re = dask.compute(*tasks)
return re

# main_file.py
from method_file import TestClass
import numpy as np
import pandas as pd 
if __name__ == '__main__':
ar = np.arange(500000000).reshape(5000000, 100)
pdf = pd.DataFrame(ar)
tc = TestClass()
tc.from_dataframe(pdf)
print(tc.do())
print(pdf.head(3))

python3 main_file.py

该输出：

(5000117, 5000054, 5000304, 5000111, 5000010, 5000264, 5000201, 5000346, 5000486, 5000376)
0    1    2    3    4    5    6   ...   93   94   95   96   97   98   99
0  9999    1    2    3    4    5    6  ...   93   94   95   96   97   98   99
1   100  101  102  103  104  105  106  ...  193  194  195  196  197  198  199
2   200  201  202  203  204  205  206  ...  293  294  295  296  297  298  299
[3 rows x 100 columns]

这意味着work方法能够读取data全局变量。不仅如此，它甚至突变了pdf变量！我知道fork的多处理也可以通过这种方式读取数据，如下所示。

2. 这里是使用fork启动方法的多处理

# method_file2.py
from multiprocessing import Pool
import multiprocessing
import random
class TestClass():
def from_dataframe(self, pdf):
global data
data = pdf
@staticmethod
def work(elem):
data.iloc[0, 0] = 9999
return len(data) + elem
def do(self):
multiprocessing.set_start_method('fork')
pool = Pool(6)
procs = [pool.apply_async(TestClass.work, args=(random.randint(1,500), )) for i in range(1, 10)]
re = [proc.get() for proc in procs]
return re

# main_file2.py
from method_file2 import TestClass
import numpy as np
import pandas as pd 
if __name__ == '__main__':
ar = np.arange(500000000).reshape(5000000, 100)
pdf = pd.DataFrame(ar)
tc = TestClass()
tc.from_dataframe(pdf)
print(tc.do())
print(pdf.head(3))

python3 main_file2.py

该输出：

[5000456, 5000346, 5000122, 5000120, 5000358, 5000067, 5000375, 5000444, 5000288]
0    1    2    3    4    5    6   ...   93   94   95   96   97   98   99
0    0    1    2    3    4    5    6  ...   93   94   95   96   97   98   99
1  100  101  102  103  104  105  106  ...  193  194  195  196  197  198  199
2  200  201  202  203  204  205  206  ...  293  294  295  296  297  298  299
[3 rows x 100 columns]

正如您所看到的，它可以读取，因为主进程的状态是复制的，但它不能更改对象(pdf(。

为了简洁起见，我不会在这里包含更多的代码，但我也尝试了定时cloudpicklepdf对象，我知道Dask并没有pickle那个数据帧。Dask怎么可能以这种方式跨进程共享内存？

Answer 1

我没有看到您设置任何进程。Dask的默认调度程序使用线程池，因此所有任务都可以看到相同的变量。看见https://docs.dask.org/en/latest/scheduler-overview.html