我们将 python Beam SDK 与 GCP 的数据流结合使用。我们的管道依赖于我们知道有一些延迟的外部系统。如何编写等待 N 分钟的管道（其中 N 是我在启动作业时提供的常量）。
类似的东西 pubsub -> (休眠1分钟) -> 从外部系统读取数据
我对“FixedWindow”的理解是将数据分组到时间范围内，所以如果我使用 60 秒的固定窗口，我可以实现“最多 60 秒”的延迟，但我希望所有传入数据的延迟为 60 秒。
【问题讨论】：
标签： google-cloud-dataflow apache-beam
由于 Window 问题是由 @Kenn Knowles 回答的，请允许我回答另一半。
我认为您可以使用Stateful and Timely processing 并为每个元素使用一分钟的Timer。
请记住，计时器适用于每个键，因此每个键都必须是唯一的才能使其工作。我制作了这个代码示例，以便您可以测试它，阅读主题 projects/pubsub-public-data/topics/taxirides-realtime。
p
.apply("Read From PubSub", PubsubIO.readStrings().fromTopic(options.getTopic()))
.apply("Parse and to KV", ParDo.of(new DoFn<String, KV<String, String>>() {
@ProcessElement
public void processElement(ProcessContext c) throws ParseException {
JSONObject json = new JSONObject(c.element());
String rideStatus = json.getString("ride_status");
// ride_id is unique for dropoff
String rideId = json.getString("ride_id"); // this is the session
if (rideStatus.equals("dropoff")) {
c.output(KV.of(rideId, "value"));
}
}
}
))
// Stateful DoFn need to have a KV as input
.apply("Timer", ParDo.of(new DoFn<KV<String, String>, String >() {
private final Duration BUFFER_TIME = Duration.standardSeconds(60);
@TimerId("timer")
private final TimerSpec timerSpec = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
@StateId("buffer")
// Elements will be saved here, with type String
private final StateSpec<BagState<String>> bufferedEvents = StateSpecs.bag();
@ProcessElement
public void processElement(ProcessContext c,
@TimerId("timer") Timer timer,
@StateId("buffer") BagState<String> buffer)
throws ParseException {
// keys are unique, so no need to use counters to trigger the offset
timer.offset(BUFFER_TIME).setRelative();
buffer.add(c.element().getKey()); // add to buffer the unique id
LOG.info("TIMER: Adding " + c.element().getKey() +
" buffer at " + Instant.now().toString());
}
// This method is call when timers expire
@OnTimer("timer")
public void onTimer(
OnTimerContext c,
@StateId("buffer") BagState<String> buffer
) throws IOException {
for (String id : buffer.read()) { // there should be only one since key is unique
LOG.info("TIMER: Releasing " + id +
" from buffer at " + Instant.now().toString());
c.output(id);
}
buffer.clear(); // clearing buffer
}
})
);
这只是一个快速测试，所以代码中可能会有一些需要改进的地方。
不过，我不确定这将如何处理大量元素，因为您在单个计时器中将所有元素缓存一分钟。我目前正在 Dataflow 中运行这个管道，到目前为止一切顺利，如果发生奇怪的事情，我会更新它。
这与使用sleeps 相比的优势在于sleep 需要等待包中的每个元素进入休眠状态，而这会并行等待。缺点可能是使用了过多的 shuffle，但我没有对此进行过多的测试来确定这一点。
请注意，在“正常”有状态 DoFns 中，(1) 键不应该是唯一的，在这种情况下，bag 将添加多个元素，(2) 使用计数器或其他东西来了解是否已经需要计时器已偏移，在这种情况下我们不需要它，因为键是唯一的
这里有管道工作的屏幕截图
【讨论】：
FixedWindows 不会引入任何延迟。
在 Beam 中，窗口化根据元素的时间戳对元素进行分组。这与元素到达的时间是分开的。
PubsubIO 转换维护一个“水印”，用于测量仍保留在 Pubsub 队列中的时间戳。所以水印会实时滞后1分钟。
如果发布订阅主题长时间为空，水印会实时同步。因此，在这种情况下，您可能需要在管道中允许延迟数据。
【讨论】：