?问题的提出

? 对于WaterMark设计的位置是否会影响窗口的正常开闭?

??下面我模拟了两种情景(source并行度为1,map并行度为2),分别是
1.在source后设置watermark,经过map后开窗
2.在map后设置watermark,然后开窗

ps:? ? ? ?下面的两种代码我都设置了自然增长的watermark,窗口时间都是5秒,只是设置watermark的位置不同

????????????????watermark是testWM对象的ts字段*1000

?代码一:在Source后添加WaterMark

public class WMTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        // TODO: 2021/12/1 在source后设置watermark 
        SingleOutputStreamOperator<String> sourceWithWM = source.assignTimestampsAndWatermarks(WatermarkStrategy
                .<String>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        String[] split = element.split(",");
                        return Long.parseLong(split[2]) * 1000;
                    }
                }));
        // TODO: 2021/12/1 设置map并行度为2 
        SingleOutputStreamOperator<testWM> mapDS = sourceWithWM.map(r -> {
            String[] split = r.split(",");
            return new testWM(Integer.parseInt(split[0]), Integer.parseInt(split[1]),Long.parseLong(split[2]));
        }).setParallelism(2);
        SingleOutputStreamOperator<String> resultDS = mapDS.keyBy(r -> r.getId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<testWM, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, Context context, Iterable<testWM> elements, Collector<String> out) throws Exception {
                        out.collect("我关窗啦");
                    }
                });
        resultDS.print();
        env.execute();

    }
}
@Data
@AllArgsConstructor
class testWM{
    private int id;
    private int num;
    private long ts;
}

?代码二:在Map后设置WaterMark

public class WMTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        // TODO: 2021/12/1 设置map并行度为2
        SingleOutputStreamOperator<testWM> mapDS = source.map(r -> {
            String[] split = r.split(",");
            return new testWM(Integer.parseInt(split[0]), Integer.parseInt(split[1]),Long.parseLong(split[2]));
        }).setParallelism(2);
        // TODO: 2021/12/1  在map后添加watermark
        SingleOutputStreamOperator<testWM> mapWithWM = mapDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<testWM>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<testWM>() {
                    @Override
                    public long extractTimestamp(testWM element, long recordTimestamp) {
                        return element.getTs() * 1000;
                    }
                }));
        SingleOutputStreamOperator<String> resultDS = mapWithWM.keyBy(r -> r.getId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<testWM, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, Context context, Iterable<testWM> elements, Collector<String> out) throws Exception {
                        out.collect("我关窗啦");
                    }
                });
        resultDS.print();
        env.execute();

    }
}
@Data
@AllArgsConstructor
class testWM{
    private int id;
    private int num;
    private long ts;
}