在 Dask 数据框中滞后值的最佳方法是什么？

如何解决在 Dask 数据框中滞后值的最佳方法是什么？

我有一个名为 data 的 Dask 数据帧，它非常大，无法放入主内存中，而且重要的是未排序。数据帧在以下键上是唯一的：[strike,expiration,type,time]。我需要在 Dask 中完成的内容相当于 Pandas 中的以下内容：

data1 = data[['strike','expiration','type','time','value']].sort_values()
data1['lag_value'] = data1.groupby(['strike','time'])['value'].shift(1)

换句话说，我需要在一个分组中滞后变量 value。在 Dask 中执行此操作的最佳方法是什么 - 我知道排序在计算上会非常昂贵，但鉴于我想做的事情，我认为没有办法解决它？

先谢谢你！

解决方法

我会做一些假设，但我的猜测是数据是“有点”排序的。因此，如果您正在处理高频数据，您可能有特定于一天或一周或一小时的文件分区。这意味着您可以在这些分区内进行排序，这通常是一项更易于管理的任务。

如果这个猜测是错误的，那么产生排序（和持久化）数据的固定成本可能是个好主意，因为它会加速你的下游分析。

由于您只有一个大文件并且它不是很大（如果您可以访问集群，25GB 应该可以管理），最好的办法可能是使用常规 Pandas 加载到内存中，使用分区对数据进行排序和保存日期/到期日/行情（如果有）或其他一些对您的下游分析有意义的列划分。

通过使用适当的 dtypes 可能会减少内存占用，例如，strike、type、expiration 列可能会占用更少的空间作为类别（与字符串相比）。

如果根本无法一次性将其加载到内存中，则可以使用 Pandas 对行块进行迭代，然后将相关位保存在较小的块中，这是粗略的伪代码：

df = pd.read_csv('big_file',iterator=True,chunksize=10**4)
for rows in df:
    # here we want to split into smaller sets based on some logic
    # note the mode is append so some additional check on file
    # existence should be added
    for group_label,group_df in rows.groupby(['type','strike']):
        group_df.to_csv(f"{group_label}.csv",mode='a')

现在上面的内容可能听起来很奇怪，因为问题被标记为 dask 而我关注的是 pandas，但我们的想法是通过对相关变量的数据进行分区来节省下游时间。使用 dask 也可能实现，但根据我在此类情况下的经验，由于工作人员之间的数据改组，我会遇到内存问题。当然，如果在您的情况下有很多文件而不是一个文件，那么使用 dask.delayed 进行一些并行化会有所帮助。

现在， 对数据进行分区/索引之后，dask 在对许多较小的块进行操作时会很好地工作。例如，如果您根据日期对数据进行分区并且您的下游分析主要使用日期，那么像 groupby 和 shift 这样的操作将非常快，因为工作人员不需要相互检查是否它们具有重叠的日期，因此大多数处理将在分区内进行。