我正在使用 X 大小和 Y 周期的滑动时间窗口。为了标记每个窗口的输出,我想获取PCollection当前窗口的时间戳。
PCollection<T> windowedInput = input
.apply(Window<T>into(
SlidingWindows.of(Duration.standardMinutes(10))
.every(Duration.standardMinutes(1))));
// Extract key from each input and run a function per group.
//
// Q: ExtractKey() depends on the window triggered time.
// How can I pass the timestamp of windowedInputs to ExtractKey()?
PCollection<KV<K, Iterable<T>>> groupedInputs = windowedInputs
.apply(ParDo.of(new ExtractKey()))
.apply(GroupByKey.<K, Ts>create());
// Run Story clustering and write outputs.
//
// Q: Also I'd like to add a window timestamp suffix to the output.
// How can I pass (or get) the timestamp to SomeDoFn()?
PCollection<String> results = groupedInputs.apply(ParDo.of(new SomeDoFn()));
A DoFn
允许通过可选的访问当前元素的窗口BoundedWindow
上的参数@ProcessElement
method:
class SomeDoFn extends DoFn<KV<K, Iterable<T>>, String> {
@ProcessElement
public void process(ProcessContext c, BoundedWindow window) {
...
}
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)