在 python 中处理时间序列数据：对一个时间段的序列求和和聚合

Question

我试图弄清楚如何可视化一些传感器数据。我每 5 分钟为多个设备收集一次数据，存储在一个类似这样的 JSON 结构中（请注意，我无法控制数据结构）：

[
  {
    "group": { "id": "01234" },
    "measures": {
      "measures": {
        "...device 1 uuid...": {
          "metric.name.here": {
            "mean": [
              ["2019-04-17T14:30:00+00:00", 300, 1],
              ["2019-04-17T14:35:00+00:00", 300, 2],
              ...
            ]
          }
        },
        "...device 2 uuid...": {
          "metric.name.here": {
            "mean": [
              ["2019-04-17T14:30:00+00:00", 300, 0],
              ["2019-04-17T14:35:00+00:00", 300, 1],
              ...
            ]
          }
        }
      }
    }
  }
]

["2019-04-17T14:30:00+00:00", 300, 0] 形式的每个元组都是[timestamp, granularity, value]。设备按项目 ID 分组。在任何给定的组中，我想获取多个设备的数据并将它们汇总在一起。例如，对于上述示例数据，我希望最终系列看起来像：

["2019-04-17T14:30:00+00:00", 300, 1],
["2019-04-17T14:35:00+00:00", 300, 3],

系列不一定是相同的长度。

最后，我想将这些测量结果汇总为每小时样本。

我可以像这样获得单个系列：

with open('data.json') as fd:
  data = pd.read_json(fd)

for i, group in enumerate(data.group):
    project = group['project_id']
    instances = data.measures[i]['measures']
    series_for_group = []
    for instance in instances.keys():
        measures = instances[instance][metric][aggregate]

        # build an index from the timestamps
        index = pd.DatetimeIndex(measure[0] for measure in measures)

        # extract values from the data and link it to the index
        series = pd.Series((measure[2] for measure in measures),
                           index=index)

        series_for_group.append(series)

在外部for 循环的底部，我有一个pandas.core.series.Series 对象数组，代表与当前组相关的不同测量集。我希望我可以像total = sum(series_for_group) 那样简单地将它们加在一起，但这会产生无效数据。

1. 我是否正确读取了这些数据？这是我第一次与 Pandas 合作；我不确定 (a) 创建一个索引，然后 (b) 填充数据是否是正确的过程。

2. 如何成功地将这些系列相加？

3. 如何将此数据重新采样为 1 小时间隔？查看this question，看起来.groupby 和.agg 方法似乎很有趣，但从该示例中不清楚如何指定间隔大小。

更新 1

也许我可以使用concat 和groupby？例如：

final = pd.concat(all_series).groupby(level=0).sum()

Answer 1

我在评论中建议做这样的事情：

result = pd.DataFrame({}, columns=['timestamp', 'granularity', 'value',
                               'project', 'uuid', 'metric', 'agg'])
for i, group in enumerate(data.group):
    project = group['id']
    instances = data.measures[i]['measures']

    series_for_group = []


    for device, measures in instances.items():
        for metric, aggs in measures.items():
            for agg, lst in aggs.items():
                sub_df = pd.DataFrame(lst, columns = ['timestamp', 'granularity', 'value'])
                sub_df['project'] = project
                sub_df['uuid'] = device
                sub_df['metric'] = metric
                sub_df['agg'] = agg

                result = pd.concat((result,sub_df), sort=True)

# parse date:
result['timestamp'] = pd.to_datetime(result['timestamp'])

这导致数据看起来像这样

    agg     granularity         metric  project     timestamp           uuid                value
0   mean    300     metric.name.here    01234   2019-04-17 14:30:00     ...device 1 uuid...     1
1   mean    300     metric.name.here    01234   2019-04-17 14:35:00     ...device 1 uuid...     2
0   mean    300     metric.name.here    01234   2019-04-17 14:30:00     ...device 2 uuid...     0
1   mean    300     metric.name.here    01234   2019-04-17 14:35:00     ...device 2 uuid...     1

然后你可以做整体聚合

result.resample('H', on='timestamp').sum()

给出：

timestamp
2019-04-17 14:00:00    4
Freq: H, Name: value, dtype: int64

或groupby聚合：

result.groupby('uuid').resample('H', on='timestamp').value.sum()

给出：

uuid                 timestamp          
...device 1 uuid...  2019-04-17 14:00:00    3
...device 2 uuid...  2019-04-17 14:00:00    1
Name: value, dtype: int64

Answer 2

要从具有不同长度的系列（例如 s1、s2、s3）构建数据帧 (df)，您可以尝试：

df=pd.concat([s1,s2,s3], ignore_index=True, axis=1).fillna('')

一旦你构建了你的数据框：

1. 确保所有日期都存储为时间戳对象：

   df['Date']=pd.to_datetime(df['Date'])

然后，添加另一列以从日期列中提取小时数：

df['Hour']=df['Date'].dt.hour

然后按小时分组并总结值：

df.groupby('Hour').sum()

Answer 3

根据我的问题中的代码，我最终得到了一个看似可行的解决方案。在我的系统上，处理大约 85MB 的输入数据大约需要 6 秒。相比之下，我在 5 分钟后取消了 Quang 的代码。

我不知道这是否是处理这些数据的正确方法，但它会产生明显正确的结果。我注意到在这个解决方案中构建一个系列列表，然后进行单个 pd.concat 调用比将 pd.concat 放入循环中更高效。

#!/usr/bin/python3

import click
import matplotlib.pyplot as plt
import pandas as pd


@click.command()
@click.option('-a', '--aggregate', default='mean')
@click.option('-p', '--projects')
@click.option('-r', '--resample')
@click.option('-o', '--output')
@click.argument('metric')
@click.argument('datafile', type=click.File(mode='rb'))
def plot_metric(aggregate, projects, output, resample, metric, datafile):

    # Read in a list of project id -> project name mappings, then
    # convert it to a dictionary.
    if projects:
        _projects = pd.read_json(projects)
        projects = {_projects.ID[n]: _projects.Name[n].lstrip('_')
                    for n in range(len(_projects))}
    else:
        projects = {}

    data = pd.read_json(datafile)
    df = pd.DataFrame()

    for i, group in enumerate(data.group):
        project = group['project_id']
        project = projects.get(project, project)

        devices = data.measures[i]['measures']
        all_series = []
        for device, measures in devices.items():
            samples = measures[metric][aggregate]
            index = pd.DatetimeIndex(sample[0] for sample in samples)
            series = pd.Series((sample[2] for sample in samples),
                               index=index)
            all_series.append(series)

        # concatenate all the measurements for this project, then
        # group them using the timestamp and sum the values.
        final = pd.concat(all_series).groupby(level=0).sum()

        # resample the data if requested
        if resample:
            final = final.resample(resample).sum()

        # add series to dataframe
        df[project] = final

    fig, ax = plt.subplots()
    df.plot(ax=ax, figsize=(11, 8.5))
    ax.legend(frameon=False, loc='upper right', ncol=3)

    if output:
        plt.savefig(output)
        plt.close()
    else:
        plt.show()


if __name__ == '__main__':
    plot_metric()