如何在apachebeam中创建基于元组的滑动窗口？这在Flink很容易做到：
DataStream.countWindowAll(long size, long slide)
但beam（或dataflow）的文档还不清楚如何做到这一点。是窗口和触发器的组合吗？效率高吗？
kgsdhlau1#
滑动窗本身由梁支撑。请参阅slidingwindows类的编程指南和文档。例如。：
PCollection<Foo> foos = ...;PCollection<Integer> counts = foos .apply(Window.into( SlidingWindows.of(Duration.standardMinutes(5)) .every(Duration.standardMinutes(1)))) // Below is required instead of Count.globally() when you use // a non-global windowing function. .apply(Combine.globally(Count.<Foo>combineFn()).withoutDefaults());PCollection<String> formattedCounts = counts.apply( ParDo.of(new DoFn<Integer, String>() { @ProcessElement public void process(ProcessContext c, BoundedWindow w) { c.output("Window: " + w + ", count: " + c.element()); } }));
PCollection<Foo> foos = ...;
PCollection<Integer> counts = foos
.apply(Window.into(
SlidingWindows.of(Duration.standardMinutes(5))
.every(Duration.standardMinutes(1))))
// Below is required instead of Count.globally() when you use
// a non-global windowing function.
.apply(Combine.globally(Count.<Foo>combineFn()).withoutDefaults());
PCollection<String> formattedCounts = counts.apply(
ParDo.of(new DoFn<Integer, String>() {
@ProcessElement
public void process(ProcessContext c, BoundedWindow w) {
c.output("Window: " + w + ", count: " + c.element());
}
}));
触发是问题的一个独立维度，它控制特定窗口的数据何时被认为“足够完整”以应用聚合。请参阅编程指南。
1条答案
按热度按时间kgsdhlau1#
滑动窗本身由梁支撑。请参阅slidingwindows类的编程指南和文档。
例如。：
触发是问题的一个独立维度，它控制特定窗口的数据何时被认为“足够完整”以应用聚合。请参阅编程指南。