python - 如何使用 TimeGrouper 迭代包含不同范围的多个文件

Question

我有一个文件集合。每个文件有 1 秒数据。此外，这些文件不是周期性的，即它们不是每日文件。例如，一个文件可能包含一天半的数据，而下一个文件可能包含3天2小时的数据；文件之间和文件内部可能存在间隙。另一个问题是同时加载内存中的所有文件是不切实际的。

这是一个显示问题的具体示例。以下数据帧有一天半的 1 秒数据:

index = pd.date_range('now', periods=60*60*24*1.5, freq='1S')
data_a = pd.DataFrame(np.random.rand(len(index)), index=index, columns=['data'])

下一个数据帧从上一个数据帧停止的地方开始，它有两天的数据:

index = pd.date_range(data_a.index[-1] + pd.Timedelta('1S'), periods=60*60*24*2, freq='1S')
data_b = pd.DataFrame(np.random.rand(len(index)), index=index, columns=['data'])

让我们在每个数据帧上创建 10 分钟的迭代器和 chain他们:

ia = iter(data_a.groupby(TimeGrouper('10Min')))
ib = iter(data_b.groupby(TimeGrouper('10Min')))
iaib = chain(ia, ib)

如果我们迭代 iaib，期望的行为是仅查看每个组键(及其数据)一次，但事实并非如此。

seen = {}
for name, group in iaib:
    count = seen.get(name, 0)
    seen[name] = count + 1

seen_twice = {key: value for key, value in seen.items() if value > 1}

seen_twice的内容是:

{Timestamp('2017-06-02 08:50:00', freq='10T'): 2}

在此示例中，2017-06-02 08:50:00 是最后一组 data_a 和第一组 data_b< 的键.

如何在所有文件上按 10 分钟组进行迭代，而不在文件边缘重复组？

Answer 1

该解决方案分为两部分:一是将所有文件作为单个数据集处理；二是将所有文件作为一个数据集进行处理。另一个是考虑到这样一个事实:10 分钟的组可以分为一个文件的结尾和下一个文件的开始。

这些是所需的导入:

from itertools import chain

import pandas as pd
from pandas.tseries.resample import TimeGrouper

将所有文件作为单个数据集处理

此函数返回给定文件的 10 分钟组的迭代器:

def make_iterator(file):
    df = pd.read_csv(file, index_col='timestamp', parse_dates=['timestamp'])
    return iter(df.groupby(TimeGrouper('10Min')))

上面的函数用于创建 itertools.chain 的迭代器的迭代器。给定一个文件列表，可以像这样创建文件集合的所有 10 分钟组上的单个迭代器:

files = ... # list obtained by os.listdir() or glob.glob()    
iterator_of_single_file_group_iterators = map(make_iterator, files)
chained_file_group_iterator = chain.from_iterable(iterator_of_single_file_group_iterators)

考虑到一个组可以在一个文件的末尾和下一个文件的开头之间分割的事实

但是，上面的迭代器不知道跨两个文件的 10 分钟组。下面的类解决了这个问题:

class TimeGrouperChainDecorator(object):

    def __init__(self, iterator):
        self.iterator = iterator
        self._has_more = True
        self._last_item = next(self.iterator)

    def __iter__(self):
        return self

    def __next__(self):
        if not self._has_more:
            raise StopIteration
        try:
            return self._next()
        except StopIteration:
            self._has_more = False
            if self._last_item is not None:
                return self._last_item
            raise StopIteration

    def _next(self):
        new_key, new_data = next(self.iterator)

        last_key, last_data = self._last_item
        if new_key == last_key:
            data = pd.concat([last_data, new_data])
            try:
                self._last_item = next(self.iterator)
            except StopIteration:
                self._has_more = False
            return new_key, data
        else:
            self._last_item = new_key, new_data
            return last_key, last_data

请注意，该实现完全依赖于 pandas groupby API。要使用它，请使用上述链接的迭代器创建该类的实例:

iterator = TimeGrouperChainDecorator(chained_file_group_iterator)

for name, group in iterator:
    # do something with each 10 minute group

我的实现可能并不完美，因此欢迎任何反馈。我发布了snippet with 3 tests .