我是Dask新手,将熊猫数据框导出到具有行组的拼花地板:
x.to_parquet(path + 'ohlcv_TRX-PERP_978627_rowgrouped.prq', row_group_size=1000)
然后我尝试用Dask加载它,它似乎工作正常(?):
x = dd.read_parquet(path + 'ohlcv_TRX-PERP_978627_rowgrouped.prq')
x
# Note: The dataframe has almost 2000 columns, I clipped them for here
Dask DataFrame Structure:
open h
npartitions=978
2019-07-21 23:55:00 float64 floa
2019-07-22 16:35:00 ...
... ...
2021-05-30 17:06:00 ...
2021-05-31 03:32:00 ...
Dask Name: read-parquet, 978 tasks
到目前为止,没有问题。但是当我调用x.max().compute()
时,Dask似乎将整个数据集加载到RAM中(至少RAM会疯狂地加速),然后崩溃。只看max()
:
x = x.max()
x
Dask Series Structure:
npartitions=1
ACCBL_10 float64
volume ...
dtype: float64
Dask Name: dataframe-max-agg, 1957 tasks
根据Dask教程https://tutorial.dask.org/04_dataframe.html#Computations-with-dask.dataframe据我的理解,这应该可以很好地工作,尽管(?)
当我尝试仅对一列调用max()
时,它也会耗尽内存:
x.open.max().compute()
我是做错了什么,还是应该这样做,我必须做些不同的事情
我现在还尝试使用distributed
系统,并将客户端限制为10GB,但Dask再次占用24GB的RAM,只是打印了一条警告,表示工作组完全超出了设置的内存限制:
if __name__ == '__main__':
client = Client(processes=False, memory_limit='5GB')
x = dd.read_parquet(path + 'ohlcv_TRX-PERP_978627_rowgrouped.prq')
print(x)
s = x.max().compute()
print(s)
distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 24.13 GB -- Worker memory limit: 5.00 GB
如果可能的话,我会将拼花地板保存到多个文件中(大小取决于您的硬件,但在笔记本电脑上,每个分区大约100-200 MB就可以了)。如果这不是一个选项,请尝试以下操作:
这样做的目的是要求
dask
计算每个分区的最大值,然后比较每2个分区的最大值,这样可以减少内存占用,但代价是要运行更多的任务。您可以使用split_every
数字来查看硬件上是否可以容忍更高的值,但希望2可以工作另外,如果您打算使用单个文件,那么使用
vaex
可能会获得更好的性能,请参见this comparison相关问题 更多 >
编程相关推荐