将带有 start - end 的行转换为带有 TimeIndex 的数据帧的性能问题

Posted 2023-02-23

【中文标题】将带有 start - end 的行转换为带有 TimeIndex 的数据帧的性能问题

【英文标题】：Performance issue turning rows with start - end into a dataframe with TimeIndex

【发布时间】：2018-11-17 17:32:19

【问题描述】：

我有一个大型数据集，其中每一行代表某个时间间隔（开始和结束之间）的某种类型（想想传感器）的值。 它看起来像这样：

    start       end    type value
2015-01-01  2015-01-05  1   3
2015-01-06  2015-01-08  1   2
2015-01-05  2015-01-08  3   3
2015-01-13  2015-01-16  2   1

我想把它变成这样的每日时间索引框架：

day       type  value
2015-01-01  1   3
2015-01-02  1   3
2015-01-03  1   3
2015-01-04  1   3
2015-01-05  1   3
2015-01-06  1   2
2015-01-07  1   2
2015-01-08  1   2
2015-01-05  3   3
2015-01-16  3   3
2015-01-07  3   3
2015-01-08  3   3
2015-01-13  2   1
2015-01-14  2   1
2015-01-15  2   1
2015-01-16  2   1

（请注意，我们不能对间隔做出任何假设：它们应该是连续的且不重叠，但我们不能保证）

基于这些 Stack Overflow 答案 [1] (DataFrame resample on date ranges) [2] (pandas: Aggregate based on start/end date)，似乎存在两种方法：一种围绕 itertuples，一种围绕 melt（2 上面使用了 stack/unstack，但它是类似于融化）。 让我们比较它们的性能。

# Creating a big enough dataframe
date_range = pd.date_range(start=dt.datetime(2015,1,1), end=dt.datetime(2019,12,31), freq='4D')
to_concat = []
for val in range(1,50):
    frame_tmp = pd.DataFrame()
    frame_tmp['start'] = date_range
    frame_tmp['end'] = frame_tmp['start']+ dt.timedelta(3)
    frame_tmp['type'] = val
    frame_tmp['value'] = np.random.randint(1, 6, frame_tmp.shape[0])
    to_concat.append(frame_tmp)
df = pd.concat(to_concat, ignore_index=True)

# Method 1 
def method_1(df):
    df1 = (pd.concat([pd.Series(r.Index,
                                pd.date_range(r.start,
                                              r.end,
                                              freq='D'))
                      for r in df.itertuples()])) \
        .reset_index()
    df1.columns = ['start_2', 'idx']

    df2 = df1.set_index('idx').join(df).reset_index(drop=True)

    return df2.set_index('start_2')

df_method_1=df.groupby(['type']).apply(method_1)

# Method 2
df_tmp= df.reset_index()
df1 = (df_tmp.melt(df_tmp.columns.difference(['start','end']),
          ['start', 'end'],
          value_name='current_time')
  )
df_method_2 = df1.set_index('current_time').groupby('index', group_keys=False)\
.resample('D').ffill()

对于 Jupyter 中的 %%timeit，方法 1 需要约 8 秒，而方法 2 需要约 25 秒来定义作为示例的数据帧。这太慢了，因为我正在处理的真实数据集比这大得多。在该数据帧上，方法 1 大约需要 20 分钟。

您知道如何加快速度吗？

【问题讨论】：

我无法运行您的代码，frame_tmp['start'] = date_range 中的date_range 是什么？

好收获！我已经更新了代码。谢谢@Ben.T

什么是df_tmp？此外，由于您说“我们不能对间隔做出任何假设”，我假设您想要扩展指定的读数（即使它们重叠或有间隙），而不是在可用读数中查找一年中的每一天。对吗？

好收获！只是一个 .reset_index()。我已经更新了代码。是的，我想扩展特定的读数，因为这将允许我以通用方式进行任何类型的检查。 @MatthiasFripp

【参考方案1】：

这比你的 method_1 快大约 1.7 倍，而且更整洁：

df_expand = pd.DataFrame.from_records(
    (
        (d, r.type, r.value) 
        for r in df.itertuples()
        for d in pd.date_range(start=r.start, end=r.end, freq='D')
    ),
    columns=['day', 'type', 'row']
)

通过创建自己的日期范围而不是调用pd.date_range()，您可以将速度提高约 7 倍：

one_day = dt.timedelta(1)
df_expand = pd.DataFrame.from_records(
    (
        (r.start + i * one_day, r.type, r.value) 
        for r in df.itertuples()
        for i in range(int((r.end-r.start)/one_day)+1)
    ),
    columns=['day', 'type', 'row']
)

或者使用 numpy 的 arange 函数生成日期，您可以将速度提高 24 倍：

one_day = dt.timedelta(1)
df_expand = pd.DataFrame.from_records(
    (
        (d, r.type, r.value) 
        for r in df.itertuples()
        for d in np.arange(r.start.date(), r.end.date()+one_day, dtype='datetime64[D]')
    ),
    columns=['day', 'type', 'row']
)

我忍不住又增加了一个，速度比上一个快两倍多一点。不幸的是，它更难阅读。这会根据读数跨越的天数（“dur”）对读数进行分组，然后使用矢量化的 numpy 操作在单个批次中扩展每个组。

def expand_group(g):
    dur = g.dur.iloc[0] # how many days for each reading in this group?
    return pd.DataFrame(
        'day': (g.start.values[:,None] + np.timedelta64(1, 'D') * np.arange(dur)).ravel(),
        'type': np.repeat(g.type.values, dur),
        'value': np.repeat(g.value.values, dur),
    )
# take all readings with the same duration and process them together using vectorized code
df_expand = (
    df.assign(dur=(df['end']-df['start']).dt.days + 1)
    .groupby('dur').apply(expand_group)
    .reset_index('dur', drop=True)
)

更新：回应您的评论，下面是矢量化方法的简化版本，它更快且更易于阅读。这不是使用groupby 步骤，而是使单个矩阵与最长读数一样宽，然后过滤掉不需要的条目。除非您的读数的最大持续时间比平均值长得多，否则这应该非常有效。使用测试数据帧（所有读数持续 4 天），这比 groupby 解决方案快约 15 倍，比 method_1 快约 700 倍。

dur = (df['end']-df['start']).max().days + 1
df_expand = pd.DataFrame(
    'day': (
        df['start'].values[:,None] + np.timedelta64(1, 'D') * np.arange(dur)
    ).ravel(),
    'type': np.repeat(df['type'].values, dur),
    'value': np.repeat(df['value'].values, dur),
    'end': np.repeat(df['end'].values, dur),
)
df_expand = df_expand.loc[df_expand['day']<=df_expand['end'], 'day':'value']

【讨论】：

这是一个令人难以置信的加速！我检查了激发这个问题的真实数据集。该过程的这一部分现在需要不到 25 分钟，而过去它的时间远高于 20 分钟。非常感谢！

@Phik，只是为了让你知道，在我的第三种方法中，我最初没有考虑到 np.arange() 不包含上端的事实，所以它省略了最后一个日期每次阅读。我现在已经修好了。

实际上在我的真实数据集上，您的矢量化方法（#4）比 #3 快 6.5 倍！这是令人难以置信！再次，谢谢你。我仍然想知道是否可以不使用 groupby，您是否尝试过这样的事情？

@Phik，numpy 向量化代码的诀窍在于，它在测量窗口中构造了一个矩阵，每个开始日一行，每一天一列。这只有在每个批次具有相同长度的测量窗口（以制作规则矩阵）时才有可能，这就是它按dur 分组的原因。但也许可以使用最大持续时间并为每个读数填写超出 dur 的空值，然后再删除它们。或者，您可以重复为所有行添加 dur >10、>9、>8 等条目，然后将它们重新排序在一起。但这可能会比groupby 慢。

@Phik，你说得对，去掉groupby 可以节省大量时间，同时也让代码更容易理解。请参阅上面的新版本。