缓冲和刷新Apache Beam流数据

Question

我有一个流媒体作业,初始运行必须处理大量数据.其中一个DoFn调用支持批量请求的远程服务,因此在处理有界集合时,我使用以下方法:

  private static final class Function extends DoFn<String, Void> implements Serializable {
    private static final long serialVersionUID = 2417984990958377700L;

    private static final int LIMIT = 500;

    private transient Queue<String> buffered;

    @StartBundle
    public void startBundle(Context context) throws Exception {
      buffered = new LinkedList<>();
    }

    @ProcessElement
    public void processElement(ProcessContext context) throws Exception {
      buffered.add(context.element());

      if (buffered.size() > LIMIT) {
        flush();
      }
    }

    @FinishBundle
    public void finishBundle(Context c) throws Exception {
      // process remaining
      flush();
    }

    private void flush() {
      // build batch request
      while (!buffered.isEmpty()) {
        buffered.poll();
        // do something
      }
    }
  }

有没有办法窗口数据,所以可以在无界集合上使用相同的方法？

我试过以下:

pipeline
    .apply("Read", Read.from(source))
    .apply(WithTimestamps.of(input -> Instant.now()))
    .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2L))))
    .apply("Process", ParDo.of(new Function()));

但startBundle和finishBundle被称为对每一个元素.是否有机会使用RxJava(2分钟窗口或100个元素包):

source
    .toFlowable(BackpressureStrategy.LATEST)
    .buffer(2, TimeUnit.MINUTES, 100) 

Answer 1

这是 per-key-and-windows state和timers新功能的典型用例。

状态在Beam 博客文章中进行了描述，而对于计时器，您必须依赖于 Javadoc。不用管 javadoc 怎么说支持他们的跑步者，真正的状态可以在 Beam 的能力矩阵中找到。

该模式与您编写的非常相似，但状态允许它与窗口一起工作，也可以跨包工作，因为它们在流中可能非常小。由于必须以某种方式对状态进行分区以保持并行性，因此您需要添加某种键。目前没有自动分片。

private static final class Function extends DoFn<KV<Key, String>, Void> implements Serializable {
  private static final long serialVersionUID = 2417984990958377700L;

  private static final int LIMIT = 500;

  @StateId("bufferedSize")
  private final StateSpec<Object, ValueState<Integer>> bufferedSizeSpec =
      StateSpecs.value(VarIntCoder.of());

  @StateId("buffered")
  private final StateSpec<Object, BagState<String>> bufferedSpec =
      StateSpecs.bag(StringUtf8Coder.of());

  @TimerId("expiry")
  private final TimerSpec expirySpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);

  @ProcessElement
  public void processElement(
      ProcessContext context,
      BoundedWindow window,
      @StateId("bufferedSize") ValueState<Integer> bufferedSizeState,
      @StateId("buffered") BagState<String> bufferedState,
      @TimerId("expiry") Timer expiryTimer) {

    int size = firstNonNull(bufferedSizeState.read(), 0);
    bufferedState.add(context.element().getValue());
    size += 1;
    bufferedSizeState.write(size);
    expiryTimer.set(window.maxTimestamp().plus(allowedLateness));

    if (size > LIMIT) {
      flush(context, bufferedState, bufferedSizeState);
    }
  }

  @OnTimer("expiry")
  public void onExpiry(
      OnTimerContext context,
      @StateId("bufferedSize") ValueState<Integer> bufferedSizeState,
      @StateId("buffered") BagState<String> bufferedState) {
    flush(context, bufferedState, bufferedSizeState);
  }

  private void flush(
      WindowedContext context,
      BagState<String> bufferedState,
      ValueState<Integer> bufferedSizeState) {
    Iterable<String> buffered = bufferedState.read();

    // build batch request from buffered
    ...

    // clear things
    bufferedState.clear();
    bufferedSizeState.clear();
  }
}

在这里做一些笔记：

• State 替换了您DoFn的实例变量，因为实例变量跨窗口没有内聚性。
• 缓冲区和大小只是根据需要而不是@StartBundle.
• 在BagState支持“盲”写入，所以并不需要成为任何读-修改-写，只是犯同样的方式将新的元素，当你输出。
• 在同一时间重复设置一个计时器就好了；它应该主要是一个 noop。
• @OnTimer("expiry")代替@FinishBundle，因为完成一个包不是每个窗口的事情，而是运行程序如何执行您的管道的工件。

综上所述，如果您正在向外部系统写入数据，那么在写入取决于窗口的写入方式之前，您可能希望将窗口具体化并重新窗口化到全局窗口中，因为“外部世界是全局窗口”。