dataframe连接和重新分区大文件，用于时间序列和相关性。

Question

我有11年的数据和记录(行)每秒钟，超过100列。它用一系列的日期时间(用Pandas to_datetime()创建)进行索引，我们需要能够在列之间进行一些相关分析，一次只能加载两个列。我们可能在较低的时间(例如48小时、1小时、月等)重新采样。在长达11年的时间里，把这些关联想象成11年。

数据目前在11个单独的拼花文件中(每年一个)，从11个.txt文件中单独生成Pandas。熊猫没有对任何这些文件进行分区。在内存中，这些拼花文件中的每一个都加载了大约20 In的内存。预期的目标机器只有16 GB，即使在11年中只加载1列也需要10 GB，因此两列也不适合。

是否有一个比使用Pandas更有效的解决方案，用于一次超过2列的相关分析？例如，使用Dask来(i)将它们连接起来，(ii)将它们重新划分到一定数量的分区，这样Dask一次就可以处理2列而不会炸掉内存？

在此post之后，我尝试了后一种解决方案，并做到了：

# Read all 11 parquet files in `data/`
df = dd.read_parquet("/blah/parquet/", engine='pyarrow')
# Export to 20 `.parquet` files
df.repartition(npartitions=20).to_parquet("/mnt/data2/SDO/AIA/parquet/combined")

但在第二步，达斯克炸毁了我的记忆，我的内核关闭。由于Dask非常关注处理比内存更大的数据，因此我感到惊讶的是，内存升级发生了。

我用Pandas重新处理了拼花文件，创建了大约20个行组(默认为每个文件只有一个组)。现在，无论将split_row_groups设置为True还是False，我都无法重新使用Dask (例如myseries = myseries.resample('48s').mean() )。我必须先在Dask系列中做compute()，才能将其作为Pandas，这似乎违背了在Dask中使用行组的目的。

在进行重采样时，我得到的结果是：

ValueError:只能使用已知的部门重采样数据，请参阅https://docs.dask.org/en/latest/dataframe-design.html#partitions获取更多信息。

当我使用默认的Pandas行为来编写只有1行组的拼花文件时，我没有这个问题。

Answer 1

默认情况下，dask.dataframe的结构更倾向于读取较小的“蜂巢”拼花文件，而不是将单个巨大的拼花文件分块成可管理的部分。来自dask.dataframe docs

默认情况下，

将作为dataframe中的一个分区单独加载每个拼花文件。这是表演性的，只要所有文件都有合理的大小。

我们建议将每一个文件加载到熊猫后的内存大小为10-250 MiB .过大的文件会导致单个工作人员过多地使用内存，而太小的文件会导致性能差，因为Dask的开销占主导地位。如果需要读取由大型文件组成的拼花数据集，则可以传递split_row_groups=True，让Dask按行分组而不是按文件进行数据分区。请注意，如果没有全局的split_row_groups=False文件，这种方法将不能像_metadata那样扩展，因为页脚需要从数据集中的每个文件中加载。

我想在这里尝试一些策略：

1. 只读取所需的列。因为您的文件是如此庞大，所以您甚至不希望dask试图加载第一个块来推断结构。您可以提供columns键dd.read_parquet，它将传递到解析引擎的各个阶段。在这种情况下，dd.read_parquet(filepath, columns=list_of_columns).

如果

1. 文件有多个行组，则可以使用dd.read_parquet参数split_row_groups=True。这将创建较小的块，每个块都小于完整的文件大小.

。

1. 如果(2)工作，您可以避免重新分区，或者如果需要，可以将分区重新分区到原来分区数的倍数(22、33等)。当从文件中读取数据时，dask不知道每个分区有多大，而且如果您指定的数字小于当前分区数的倍数，则分区行为没有很好地定义。在我运行的一些小测试中，重新分区11 -> 20将使前10个分区保持原样，并将最后一个分区拆分为剩下的10个！

如果您的文件位于磁盘上，则可以将该文件作为内存映射读取，以避免在重新分区之前加载数据。您可以通过将dd.read_parquet.传递给memory_map=True来做到这一点。

我肯定你不是唯一有这个问题的人。请让我们知道这是如何进行，并报告什么工作！