gpt4 book ai didi

java - 缓冲和刷新 Apache Beam 流数据

转载 作者:塔克拉玛干 更新时间:2023-11-03 05:20:07 25 4
gpt4 key购买 nike

我有一个流媒体作业,初始运行时必须处理大量数据。 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()));

但是 startBundlefinishBundle 会为每个元素调用。是否有机会使用 RxJava(2 分钟窗口或 100 个元素包):

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

最佳答案

这是 per-key-and-windows 新功能的典型用例 statetimers .

状态在a Beam blog post中描述,而对于计时器,您将不得不依赖 Javadoc。不要管 javadoc 中关于运行者支持他们的说法,真正的状态可以在 Beam 的 capability matrix 中找到。 .

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

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,因为完成 bundle 不是每个窗口的事情,而是运行者如何执行您的管道。

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

关于java - 缓冲和刷新 Apache Beam 流数据,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/42914198/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com