请教一下 flink stream API 代码里面怎么获取window 的起始时间 和结束时间?
TimeWindow有getStart()和getEnd()方法,参见示例代码:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.addSource(new FailingSource(numKeys,
numElementsPerKey,
numElementsPerKey / 3))
.rebalance()
.timeWindowAll(Time.of(windowSize, MILLISECONDS))
.apply(new RichAllWindowFunction<Tuple2<Long, IntType>, Tuple4<Long, Long, Long, IntType>, TimeWindow>() {
private boolean open = false;
@Override
public void open(Configuration parameters) {
assertEquals(1, getRuntimeContext().getNumberOfParallelSubtasks());
open = true;
}
@Override
public void apply(
TimeWindow window,
Iterable<Tuple2<Long, IntType>> values,
Collector<Tuple4<Long, Long, Long, IntType>> out) {
// validate that the function has been opened properly
assertTrue(open);
int sum = 0;
long key = -1;
for (Tuple2<Long, IntType> value : values) {
sum += value.f1.value;
key = value.f0;
}
out.collect(new Tuple4<>(key, window.getStart(), window.getEnd(), new IntType(sum)));
}
})
.addSink(new ValidatingSink(numKeys, numElementsPerKey / windowSize)).setParallelism(1);
env.execute("Tumbling Window Test");
TimeWindow有getStart()和getEnd()方法。
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。