java - 缓冲和刷新 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()));

但是每个元素都调用startBundleand 。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();
  }
}

在这里做一些笔记：

• 状态替换您DoFn的实例变量，因为实例变量在窗口之间没有凝聚力。
• 缓冲区和大小只是根据需要进行初始化，而不是@StartBundle.
• 支持“BagState盲”写入，因此不需要任何读取-修改-写入，只需以与输出时相同的方式提交新元素。
• 在同一时间重复设置计时器就可以了；它应该主要是一个noop。
• @OnTimer("expiry")代替@FinishBundle，因为完成捆绑不是每个窗口的事情，而是运行器如何执行管道的工件。

综上所述，如果您正在写入外部系统，也许您希望在执行写入方式取决于窗口的写入之前具体化窗口并重新窗口进入全局窗口，因为“外部世界是全局窗口”。

Answer 2

apache beam 0.6.0 的文档说 StateId 是“当前不受任何跑步者支持”。