静态展示类网站,东莞医疗网站建设报价,泰安招聘,关键词优化推广排名软件一 、 AllowedLateness API 延时关闭窗口
AllowedLateness 方法需要基于 WindowedStream 调用。AllowedLateness 需要设置一个延时时间#xff0c;注意这个时间决定了窗口真正关闭的时间#xff0c;而且是加上WaterMark的时间#xff0c;例如 WaterMark的延时时间为2s…一 、 AllowedLateness API 延时关闭窗口
AllowedLateness 方法需要基于 WindowedStream 调用。AllowedLateness 需要设置一个延时时间注意这个时间决定了窗口真正关闭的时间而且是加上WaterMark的时间例如 WaterMark的延时时间为2sAllowedLateness 的时间为2s那一个10的滚动窗口0-10这个单位窗口正常的关窗时间应该是超过12s的数据到达之后就关窗。而AllowedLateness 是在12s的基础上继续延长了2s也就是在14s的时候才真正去关闭 0-10s的窗口但是在12s的时候会触发窗口计算从12s之后到14s的数据每到达一个就会触发一次窗口计算。
二 、 OutputTag API 侧输出流
使用 OutputTag API 保证窗口关闭的数据依然可以获取窗口到达AllowedLateness 时间后将彻底关闭此时再属于该窗口范围内的数据将会流向 OutputTag 。 context.collect(new Event(A, /user, 1000L));Thread.sleep(3000);context.collect(new Event(B, /prod, 6500L));Thread.sleep(3000);context.collect(new Event(C, /cart, 4000L));Thread.sleep(3000);context.collect(new Event(D, /user, 7500L));System.out.println(窗口关闭 ~ );Thread.sleep(3000);context.collect(new Event(E, /cente, 8500L));Thread.sleep(3000);context.collect(new Event(F, /cente, 4000L));Thread.sleep(3000);context.collect(new Event(G, /cente, 9200L));Thread.sleep(3000);context.collect(new Event(H, /cente, 1000L));Thread.sleep(3000);context.collect(new Event(I, /cente, 1500L));Thread.sleep(3000);如果现在定义一个 5s的 滚动窗口WaterMark延时时间为2sAllowedLateness 延时时间为2s此时相当于是 WaterMark到达9s的时候才会关闭0-5的窗口也就是说最后两条数据会流向OutputTag . 当4000L数据到达后会再次触发一次窗口计算。
完全与预期一致。 完整代码
public class WindowOutputTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env Env.getEnv();DataStreamSourceEvent dataStreamSource env.addSource(new SourceFunctionEvent() {Overridepublic void run(SourceContextEvent context) throws Exception {context.collect(new Event(A, /user, 1000L));Thread.sleep(3000);context.collect(new Event(B, /prod, 6500L));Thread.sleep(3000);context.collect(new Event(C, /cart, 4000L));Thread.sleep(3000);context.collect(new Event(D, /user, 7500L));System.out.println(窗口关闭 ~ );Thread.sleep(3000);context.collect(new Event(E, /cente, 8500L));Thread.sleep(3000);context.collect(new Event(F, /cente, 4000L));Thread.sleep(3000);context.collect(new Event(G, /cente, 9200L));Thread.sleep(3000);context.collect(new Event(H, /cente, 1000L));Thread.sleep(3000);context.collect(new Event(I, /cente, 1500L));Thread.sleep(3000);}Overridepublic void cancel() {}});//operatorSingleOutputStreamOperatorEvent operator dataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(2))// 水位线延时2s.withTimestampAssigner(new SerializableTimestampAssignerEvent() {Overridepublic long extractTimestamp(Event event, long l) {return event.timestamp;}}));OutputTagEvent eventOutputTag new OutputTagEvent(late) {};WindowedStreamEvent, Boolean, TimeWindow windowedStream operator.keyBy(d - true).window(TumblingEventTimeWindows.of(Time.of(5, TimeUnit.SECONDS))).allowedLateness(Time.of(2, TimeUnit.SECONDS)).sideOutputLateData(eventOutputTag);SingleOutputStreamOperatorString windowAgg windowedStream.aggregate(new AggregateFunctionEvent, Long, Long() {Overridepublic Long createAccumulator() {return 0L;}Overridepublic Long add(Event event, Long acc) {return acc 1;}Overridepublic Long getResult(Long acc) {return acc;}Overridepublic Long merge(Long aLong, Long acc1) {return null;}}, new ProcessWindowFunctionLong, String, Boolean, TimeWindow() {Overridepublic void process(Boolean key, Context context, IterableLong iterable, CollectorString collector) throws Exception {long start context.window().getStart();long end context.window().getEnd();collector.collect(new Timestamp(start) ~ new Timestamp(end) iterable.iterator().next());}});windowAgg.print(窗口数据 );//获取测输出流中的延时数据DataStreamEvent sideOutput windowAgg.getSideOutput(eventOutputTag);sideOutput.print(测输出流- );env.execute();}}
