使用 Apache Beam 计算过去 24 小时的移动平均线

Question

我正在尝试计算过去 24 小时事件的移动平均值。我很难理解如何使用 Apache Beam 实现计算的“移动”部分。

我的场景如下：

1. 给定未绑定的事件流，其中每个事件都有 user 和 value 字段。
2. 在我将事件按user 分组以获取每个用户的流之后。
3. 计算过去 24 小时内每个用户 value 的平均值。如果我手动完成：

• 采用当前时间
• 查找事件时间为>= now - 24h的事件
• 对于这些事件，平均 value 字段以获得单个数字。

4. 将计算出的每用户平均值 value 汇入数据库表。

这里的动态方面是，当一个事件到期时（时钟向前滴答，事件时间变为< now - 24h），应该重新计算平均值。

我尝试了什么：

• 它不是FixedWindow，因为它在移动。我不想知道给定日期的每日平均 value 是多少。我想知道过去 24 小时内value 的当前平均值是多少。
• 由于用户随机生成事件，文档表明这可能是SessionWindow 的用例，但我对了解用户行为或查找会话不感兴趣。不确定SessionWindow 是否合适，因为它是从事件时间开始的，我正在寻找固定的时钟时间边界。

谁能解释一下这个移动平均线应该如何在 Beam 中实现？

我是 Apache Beam 的新手，刚刚开始。我查看了 Windowing 和 Trigger 文档并查看了排行榜示例。到目前为止，我还没有找到计算移动值的示例。

Answer 1

这取决于您想要输出的频率。

如果你想每小时有一个 24 小时移动平均线，你可以使用sliding windows，例如持续时间为 24 小时，周期为 1 小时。每小时您都会获得过去 24 小时内所有事件的分组。当与 CombinePerKey 操作（如 MeanCombineFn）聚合时，这将在元素进入时聚合并每小时（每个用户）发出平均值。

如果滑动窗口太多（例如，在 24 小时的时间跨度内每分钟滑动一次），滑动窗口可能会变得有点笨拙。这可以通过 state and timers 手动完成，通过将所有事件存储在状态中以及新元素何时进入和/或计时器触发计算并发出平均值（并清理旧状态以避免永远收集东西）。