SlidingWindows 用于 Apache Beam 上的慢速数据（大间隔）

Question

我正在使用 Chicago Traffic Tracker 数据集，其中每 15 分钟发布一次新数据。当有新数据可用时，它代表距离“实时”（example，查找_last_updt）10-15 分钟的记录。

例如，在 00:20，我得到时间戳为 00:10 的数据；在 00:35，我从 00:20 开始；在 00:50，我从 00:40 开始。因此，我可以“固定”获取新数据的间隔（每 15 分钟一次），尽管时间戳的间隔略有变化。

我正在尝试在 Dataflow (Apache Beam) 上使用这些数据，为此我正在使用滑动窗口。我的想法是收集和处理 4 个连续的数据点（4 x 15 分钟 = 60 分钟），理想情况下，一旦有新的数据点可用，就更新我对总和/平均值的计算。为此，我从代码开始：

PCollection<TrafficData> trafficData = input        
    .apply("MapIntoSlidingWindows", Window.<TrafficData>into(
        SlidingWindows.of(Duration.standardMinutes(60)) // (4x15)
            .every(Duration.standardMinutes(15))) .     // interval to get new data
        .triggering(AfterWatermark
                        .pastEndOfWindow()
                        .withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()))
        .withAllowedLateness(Duration.ZERO)
        .accumulatingFiredPanes());

不幸的是，当我从我的输入中接收到一个新的数据点时，我没有从我所追求的GroupByKey 获得新的（更新的）结果。

我的 SlidingWindows 有问题吗？还是我错过了什么？

Answer 1

一个问题可能是水印超出了窗口的末尾，并丢弃了所有后面的元素。您可以尝试在水印通过后几分钟：

PCollection<TrafficData> trafficData = input        
    .apply("MapIntoSlidingWindows", Window.<TrafficData>into(
        SlidingWindows.of(Duration.standardMinutes(60)) // (4x15)
            .every(Duration.standardMinutes(15))) .     // interval to get new data
        .triggering(AfterWatermark
                        .pastEndOfWindow()
                        .withEarlyFirings(AfterProcessingTime.pastFirstElementInPane())
                        .withLateFirings(AfterProcessingTime.pastFirstElementInPane()))
        .withAllowedLateness(Duration.standardMinutes(15))
        .accumulatingFiredPanes());

让我知道这是否有帮助。

Answer 2

所以@Pablo（根据我的理解）给出了正确的答案。但我有一些建议不适合发表评论。

我想问你是否需要滑动窗口？据我所知，固定窗口可以为您完成这项工作，并且在计算上也更简单。由于您正在使用累积触发的窗格，因此您不需要使用滑动窗口，因为您的下一个 DoFn 函数将已经对累积的窗格进行平均。

至于代码，我对早期和晚期触发逻辑进行了更改。我还建议增加窗口大小。由于您知道数据每 15 分钟出现一次，因此您应该在 15 分钟后关闭窗口，而不是在 15 分钟后关闭。但是您也不想选择最终会与 15 的倍数（如 20）发生冲突的窗口，因为在 60 分钟时您将遇到同样的问题。所以选择一个与 15 互质的数字，例如 19。还允许延迟输入。

    PCollection<TrafficData> trafficData = input        
        .apply("MapIntoFixedWindows", Window.<TrafficData>into(
            FixedWindows.of(Duration.standardMinutes(19)) 
                        .triggering(AfterWatermark.pastEndOfWindow()
                            // fire the moment you see an element 
                            .withEarlyFirings(AfterPane.elementCountAtLeast(1))
                            //this line is optional since you already have a past end of window and a early firing. But just in case 
                            .withLateFirings(AfterProcessingTime.pastFirstElementInPane()))
                        .withAllowedLateness(Duration.standardMinutes(60))
                        .accumulatingFiredPanes());

如果这能解决您的问题，请告诉我！

编辑

所以，我无法理解您是如何计算上述示例的，所以我使用的是通用示例。下面是一个通用的平均函数：

public class AverageFn extends CombineFn<Integer, AverageFn.Accum, Double> {
  public static class Accum {
    int sum = 0;
    int count = 0;
  }

  @Override
  public Accum createAccumulator() { return new Accum(); }

  @Override
  public Accum addInput(Accum accum, Integer input) {
      accum.sum += input;
      accum.count++;
      return accum;
  }

  @Override
  public Accum mergeAccumulators(Iterable<Accum> accums) {
    Accum merged = createAccumulator();
    for (Accum accum : accums) {
      merged.sum += accum.sum;
      merged.count += accum.count;
    }
    return merged;
  }

  @Override
  public Double extractOutput(Accum accum) {
    return ((double) accum.sum) / accum.count;
  }
}

为了运行它，您需要添加以下行：

PCollection<Double> average = trafficData.apply(Combine.globally(new AverageFn()));

由于您目前使用累积触发触发器，这将是解决解决方案的最简单的编码方式。

但是，如果您想使用丢弃火窗格窗口，则需要使用 PCollectionView 来存储上一个平均值并将其作为侧面输入传递给下一个平均值，以便跟踪值。这在编码中稍微复杂一些，但肯定会提高性能，因为每个窗口都会进行持续的工作，这与累积触发不同。

这对您生成自己的函数来丢弃火窗格窗口是否足够有意义？