22、Flink实战：window和Time（五）WaterMark延迟触发任务机制-程序员快看-教程，程序员编程资料站

此处举例说明WaterMark可能更容易理解，以EventTimeTumblingWindow为例：

如果设置窗口时间大小是10秒，窗口的边界可以是10000-19999；

WaterMark延迟2秒，那么当EventTime为21999的数据来到时，会触发计算EventTime为21999之前的所有

EventTime在10000-19999之间的数据。

即，真实要计算的时间窗口不变，部分迟到的数据会被纳入这个窗口一起计算。

如果是多并行的source,则每个分区中都要满足上述条件，才会触发计算窗口中的数据。

测试数据和代码如下: 第一个窗口结果是（b,2）,(a,3) ; 第二个窗口结果是(b,3),(a,4).

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class WaterMarkReview {
    public static void main(String[] args) throws Exception{
        //1.创建一个 flink steam 程序的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);  // 设置使用EventTime划分窗口,默认使用ProcessingTime

        //2. 测试使用socketTextStream
        DataStreamSource<String> input = env.socketTextStream("192.168.***.***", 8888);

        //3. 设置WaterMark
        SingleOutputStreamOperator<String> lines = input.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });

        //4.Transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> mapped = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] fields = s.split(",");
                return Tuple2.of(fields[1],Integer.parseInt(fields[2]));
            }
        });

        //由于Flink默认窗口时区是UTC-0，其他地区需要指定时间偏移量调整时区，如TumblingEventTimeWindows.of(Time.days(1),Time.hours(-8))
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = mapped.keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 也可传入第二个参数offset
                .sum(1);

        //5.sink
        summed.print();

        // 执行程序
        env.execute("WaterMarkReview");
    }
}