从 Parquet 加载 Dataframe 并计算 RAM 中的最大值

Question

我是 Dask 的新手，我使用行组将 pandas Dataframe 导出到 Parquet：

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

Answer 1

如果可能，我会将镶木地板保存到多个文件中（大小取决于您的硬件，但在笔记本电脑上每个分区大约 100-200 MB 比较好）。如果这不是一个选项，请尝试以下操作：

x.open.max(split_every=2).compute()

这样做的目的是让dask 计算每个分区的最大值，然后比较每 2 个分区的最大值，这样可以减少内存占用，但需要运行更多任务。您可以使用split_every 数字来查看您的硬件是否可以容忍更高的值，但希望 2 可以工作。

另外，如果您打算使用单个文件，使用vaex 可能会获得更好的性能，请参阅this comparison。