【问题标题】:What is the best way to lag a value in a Dask Dataframe?在 Dask Dataframe 中滞后值的最佳方法是什么?
【发布时间】:2021-06-04 23:37:28
【问题描述】:

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

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

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

提前谢谢你!

【问题讨论】:

    标签: sorting dask dask-distributed dask-delayed dask-dataframe


    【解决方案1】:

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

    如果这个猜测是错误的,那么产生对数据进行排序(和持久化)的固定成本可能是个好主意,因为它会加快您的下游分析。

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

    可以通过使用适当的 dtype 来减少内存占用,例如,罢工、类型、到期列可能占用更少的空间作为类别(与字符串相比)。

    如果根本无法一次将其加载到内存中,那么可以使用 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 在处理许多较小的块时会很好地工作。例如,如果您根据日期对数据进行分区,并且您的下游分析主要使用日期,那么像 groupbyshift 这样的操作将非常快,因为工作人员不需要相互检查他们是否有重叠的日期,因此大多数处理将发生在分区内。

    【讨论】:

      猜你喜欢
      • 2019-10-30
      • 2020-12-04
      • 2015-12-11
      • 1970-01-01
      • 2017-09-28
      • 2021-11-19
      • 2017-05-25
      • 2019-09-18
      • 1970-01-01
      相关资源
      最近更新 更多