使用dask.dataframe.set_index...，shuffle ='disk'

如何解决使用dask.dataframe.set_index...，shuffle ='disk'

非常感谢您提供的实用且非常有用的软件包。我不确定这是否是解决此问题的适当地点。对此表示歉意。此帖子的副本位于github。

我得到的结果对我来说是出乎意料的。在具有多个分区（也许大于节点数）的快速数据帧上使用set_index（...，shuffle ='disk'）时，我会丢失大量数据（大约丢失了一半的数据）。当我第二次重做索引时，问题变得更加严重。

当我仅使用一个节点但仍使用分布式调度程序时，相同的过程可以正常工作而不会丢失数据。

documentation指出：

与并行存储或分布式设置相比，set_index和merge / join之类的某些操作要比在一台计算机上的内存操作难得多。

是使用dask.distributed对大于内存的数据进行排序的最佳方法吗？还是在一台计算机上使用默认调度程序更好？

In [1]: import numpy as np
   ...: import dask.dataframe as dd
   ...: from dask_jobqueue import SLURMCluster
   ...: from dask.distributed import Client
   ...: cluster = SLURMCluster(
   ...:     cores=32,...:     processes=32,...:     memory="90GB"
   ...: )
   ...: cluster.scale(64)
   ...: client = Client(cluster)

In [2]: client.get_versions()['scheduler']
Out[2]:
{'host': {'python': '3.7.7.final.0','python-bits': 64,'OS': 'Linux','OS-release': '3.10.0-1127.8.2.el7.nsc1.x86_64','machine': 'x86_64','processor': 'x86_64','byteorder': 'little','LC_ALL': 'None','LANG': 'sv_SE'},'packages': {'python': '3.7.7.final.0','dask': '2.22.0','distributed': '2.22.0','msgpack': '1.0.0','cloudpickle': '1.5.0','tornado': '6.0.4','toolz': '0.10.0','numpy': '1.19.1','lz4': None,'blosc': None}}

In [2]: for n in [1e2,1e3,1e4,1e5,1.5e5,2e5,1e6]:
   ...:     n = int(n)
   ...:     pdf = dd.from_array(np.random.permutation(n).reshape(n,1))
   ...:     n_tasks = len(pdf.set_index(0,shuffle='tasks'))
   ...:     n_disk  = len(pdf.set_index(0,shuffle='disk'))
   ...:     print(n_tasks,n_disk,pdf.npartitions)
   ...:
100 100 1
1000 1000 1
10000 10000 1
100000 100000 2
150000 150000 3
200000 127770 4
1000000 499914 20

In [3]: for n in [1e2,pdf.npartitions)
   ...:
100 100 1
1000 1000 1
10000 10000 1
100000 100000 2
150000 83836 3
200000 120200 4
1000000 551936 20