动态定义apache beam窗口的大小

Question

我正在阅读PubSub的事件，目标是将它们分组到windows中。我希望每个窗口的结尾与每小时的0、15、30和45分钟一致。
由于这是一个流媒体作业，它可以随时启动，我想找到一种方法，将第一个窗口的大小与下一个窗口的大小对齐。
这将是流：

1. 开展工作
2. 定义为window_size这一时刻到下一刻钟之间的剩余时间
3. 从第一个窗口的末尾开始，设置window_size = int(15*60)（秒）

例如：

1. 开展工作
2. 现在是11点18分，所以修正一下
3. 当第一个窗口结束时，设置window_size = int(15*60)（秒）

在Google提供的一个示例中，使用窗口的管道定义如下，其中window_size是作为用户输入传递的参数：

def expand(self, pcoll):
  return (
          pcoll
          | "Window into Fixed Intervals" >> beam.WindowInto(window.FixedWindows(self.window_size))
          | "Add Key" >> beam.Map(lambda elem: (None, elem))
          | "Groupby" >> beam.GroupByKey()
          | "Abandon Key" >> beam.MapTuple(lambda _, val: val)
  )

Answer 1

您的用例非常适合Beam

首先，有一个基本的概念问题需要澄清：

• 用于窗口化的元素上的时间戳称为“事件时间”。它们是数据的一部分，描述流中的某个事件何时发生
• 启动和运行作业的时间称为“处理时间”。它不是您的数据的一部分

如果你不把这两者结合或混淆，你会更成功。Windows不会将“开始”或“结束”作为作业处理时间的一部分。Windows始终“存在”

使用15分钟的FixedWindows可以做你想做的事情。每个事件都与它所处的15分钟间隔相关联。启动作业或事件到达以进行处理时，不会影响此操作

更新：添加示例以说明：

假设您在11:18启动您的作业（如您的问题所述），并假设传入事件大约在同一时间生成。假设出现以下事件，并显示时间戳：

• A@11:01
• B@11:18
• C@11:15
• D@11:31
• E@11:29

元素将按如下方式分配给窗口：

• [11:00，11:15]中的A
• B在[11:15，11:30]中
• [11:15，11:30]中的C
• D在[11:30，11:45]中
• [11:15，11:30]中的E

请注意，窗口分配与您开始作业的时间、事件到达的时间或到达的顺序无关。实际上，您可以明天开始，或者在存档数据上重新运行，或者在顺序不接近的数据上重新运行，结果将是相同的。事件时间窗口基于数据