parquet - 使用从 parquet 文件创建的 dask 数据帧时内存使用过多

Question

我的 Parquet 文件为 800K 行 x 8.7K 列。我将其加载到 dask 数据框中:

import dask.dataframe as dd
dask_train_df = dd.read_parquet('train.parquet')
dask_train_df.info()

这会产生:

<class 'dask.dataframe.core.DataFrame'>
Columns: 8712 entries, 0 to 8711
dtypes: int8(8712)

当我尝试执行诸如 dask_train_df.head() 或 dask_train_df.loc[2:4].compute() 之类的简单操作时，我会遇到内存错误，甚至具有 17+ GB 的 RAM。

但是，如果我这样做:

import pandas as pd
train = pd.read_parquet('../input/train.parquet')
train.info()

产量:

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 800000 entries, 0 to 799999
Columns: 8712 entries, 0 to 8711
dtypes: int8(8712)
memory usage: 6.5 GB

我可以运行train.head()和train.loc[2:4]没有任何问题，因为一切都已经在内存中了。

1)所以我的问题是，为什么这些简单的操作使用 Dask Dataframe 会增加内存使用量，但当我使用 Pandas Dataframe 将所有内容加载到内存中时却可以正常工作？

我注意到 npartitions=1，并且在文档中看到 read_parquet“将 Parquet 数据目录读入 Dask.dataframe，每个分区一个文件”。就我而言，听起来我正在失去拥有多个分区的所有并行化能力，但是 Dask Dataframe 内存使用量不应该受到单个 Pandas Dataframe 内存量的限制吗？

2) 另外，还有一个附带问题:如果我想通过在 Dask Dataframe 中分区来并行化这个单个 parquet 文件，我该怎么做？我在 dd.read_parquet 签名中没有看到 block 大小参数。我也尝试使用重新分区功能，但我相信沿行分区和在 Parquet 文件中，我想沿列分区？

Answer 1

首先，我想评论一下，8712 列相当多，您会发现解析架构/元数据可能会花费大量时间，更不用说数据加载了。

当fastparquet加载数据时，它首先分配一个足够大小的数据帧，然后迭代列/ block (具有适当的开销，在这种情况下显然很小)并将值分配给分配的数据帧。

当您通过 Dask 运行计算(任何计算)时，在许多情况下，输入变量和其他中间对象的内存中可能存在任务内副本。这通常不是问题，因为整个数据集应该分为许多部分，并且为了能够处理大于内存的数据集，小的中间体的内存开销是值得付出的代价。我不确定您在哪一点收到副本，这可能值得调查和预防。

就您而言，整个数据集是一个分区。这将导致单个加载任务在一个线程中运行。您将不会获得任何并行性，并且任何中间内部副本都适用于整个数据集。您可以通过选择列来仅加载部分数据，从而创建分区并以这种方式实现并行性。然而，处理 Parquet 数据的典型方法是使用“行组”分区(即沿着索引)和多个文件，因此避免该问题的真正方法是使用已经适当分区的数据。

请注意，由于您可以直接使用 fastparquet/pandas 加载数据，因此您也可以使用 to_parquet 方法或 fastparquet 的 write 保存分区版本。功能。