具有事件数量限制的 Apache Beam 会话窗口

Question

我们需要在用户 10 分钟内没有活动时生成用户会话，但需要进行以下调整

• 我们还想在会话中的事件数达到 20k 时结束会话
• 从窗口中的第一个事件开始 4 小时后结束会话

对于 10 分钟的会话窗口 - 我们已完成以下工作：

'User Session Window' >> beam.WindowInto(window.Sessions(10 * 60),                                                    
                        timestamp_combiner=window.TimestampCombiner.OUTPUT_AT_EOW)

我需要知道如何添加其他两个条件 - 这样如果任何一个条件成功 - 我们想要发出行

Answer 1

为此，您可以基于现有的Sessions 编写自定义WindowFn，该BoundedWindow 使用跟踪元素计数的BoundedWindow 来跟踪这些额外的约束。请注意，这会增加管道的不确定性，因为发出的确切窗口集对元素出现的顺序很敏感。例如

class CountingIntervalWindow(BoundedWindow):
  def __init__(self, start, end, count):
    ...

class BoundedSessions(WindowFn):
  def assign(self, context):
    # type: (WindowFn.AssignContext) -> List[CountingIntervalWindow]
    timestamp = context.timestamp
    return [CountingIntervalWindow(timestamp, timestamp + self.gap_size, 1)]

  def get_window_coder(self):
    return PickleCoder()

  def merge(self, merge_context):
    # type: (WindowFn.MergeContext) -> None
    end = MIN_TIMESTAMP
    to_merge = []
    running_count = 0
    for w in sorted(merge_context.windows, key=lambda w: w.start):
      if to_merge:
        if (# Overlap
            end > w.start
            # Not too many elements.
            and running_count < self.count_limit
            # Not too large
            and w.end - to_merge[0].start < self.time_limit):
          to_merge.append(w)
          running_count += w.count
          if w.end > end:
            end = w.end
        else:
          if len(to_merge) > 1:
            merge_context.merge(
                to_merge,
                CountingIntervalWindow(to_merge[0].start, end, running_count))
          to_merge = [w]
          end = w.end
          running_count = w.count
      else:
        to_merge = [w]
        end = w.end
        running_count = w.count
    if len(to_merge) > 1:
      merge_context.merge(
          to_merge,
          CountingIntervalWindow(to_merge[0].start, end, running_count))