熊猫将不规则的时间序列与不同的频率对齐

Question

我有 3 个时间序列需要随着时间的推移进行比较。显然，它们需要对齐才能具有可比性。不幸的是，3 个时间序列中有 2 个是不规则的。其他 2 个范围为每 ID 和每天 2 到 500k 次观察。

初始时间序列每 300 毫秒可用一次，并且可以与其他两个时间序列连接。

但是我有两个问题：

1. 我上面介绍的这 3 个系列采用 ID, time, value 的格式，即每个组构成一个单独的时间序列
2. 制定连接条件，即假设 LEFT 和最细粒度可在一个时间窗口内连接，因为可能不完全匹配

编辑

一些虚拟数据

import pandas as pd
from datetime import datetime
import numpy as np
def make_df(frequency, valueName):
    date_rng = pd.date_range(start='2018-01-01', end='2018-01-02', freq=frequency)
    ts = pd.Series(np.random.randn(len(date_rng)), index=date_rng)
    groups = ['a', 'b', 'c', 'd', 'e']
    group_series = [groups[np.random.randint(len(groups))] for i in range(0, len(date_rng))]
    df = pd.DataFrame(ts, columns=[valueName])
    df['group'] = group_series
    return df
df_1 = make_df('ms', 'value_A')
display(df_1.head())
df_2 = make_df('H', 'value_B')
display(df_2.head())
df_3 = make_df('S', 'value_C')
display(df_3.head())

代码（都不是真正的pythonic）： 我在 SQL 中尝试了一些类似于 a JOIN b ON a.group = b.group AND time in window(some_seconds) 的非 equi-join，但是如果有多个匹配的记录，即不仅第一个而且都匹配/生成一行，这会出现问题。

此外，我确实对类似于 (spark): df.groupBy($"KEY", window($"time", "5 minutes")).sum("metric") 的数据进行分组，但这可能会非常有损。

然后我发现（pandas）Pandas aligning multiple dataframes with TimeStamp index 看起来已经很有趣了，但只能产生完全匹配。但是，当尝试使用df_2.join(df_3, how='outer', on=['group'], rsuffix='_1') 时，它不仅会在（确切）时间加入，而且group 也会失败，并出现需要pd.concat 的错误。

经过更多搜索后，我发现 (pyspark) https://github.com/twosigma/flint 在间隔内实现时间序列连接 - 但是，我在使用它时遇到问题。

Answer 1

我找不到在 pandas 中执行此操作的简单方法 - 所以我直接在 spark 中执行此操作。

Flint 是我选择的工具。最初，燧石不能在 spark 2.2 上工作，但我在这里修复：https://github.com/geoHeil/flint/commit/a2827d38e155ec8ddd4252dc62d89181f14f0c47 以下工作正常：

val left = Seq((1,1L, 0.1), (1, 2L,0.2), (3,1L,0.3), (3, 2L,0.4)).toDF("groupA", "time", "valueA")
  val right = Seq((1,1L, 11), (1, 2L,12), (3,1L,13), (3, 2L,14)).toDF("groupB", "time", "valueB")
  val leftTs = TimeSeriesRDD.fromDF(dataFrame = left)(isSorted = false, timeUnit = MILLISECONDS)
  val rightTS        = TimeSeriesRDD.fromDF(dataFrame = right)(isSorted = false, timeUnit = MILLISECONDS)

  val mergedPerGroup = leftTs.leftJoin(rightTS, tolerance = "1s")

即它对所有组执行某种笛卡尔连接：

mergedPerGroup.toDF.filter(col("groupA") === col("groupB")).show
+-------+------+------+------+------+
|   time|groupA|valueA|groupB|valueB|
+-------+------+------+------+------+
|1000000|     3|   0.3|     3|    13|
|2000000|     3|   0.4|     3|    14|

使用 distinct 删除重复项。