flink watermark介绍

/**
 * This generator generates watermarks assuming that elements arrive out of order,
 * but only to a certain degree. The latest elements for a certain timestamp t will arrive
 * at most n milliseconds after the earliest elements for timestamp t.
 */
public class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {

    private final long maxOutOfOrderness = 3500; // 3.5 seconds

    private long currentMaxTimestamp;

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        long timestamp = element.getCreationTime();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }

    @Override
    public Watermark getCurrentWatermark() {
        // return the watermark as current highest timestamp minus the out-of-orderness bound
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

/**
 * This generator generates watermarks that are lagging behind processing time by a fixed amount.
 * It assumes that elements arrive in Flink after a bounded delay.
 */
public class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {

    private final long maxTimeLag = 5000; // 5 seconds

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        return element.getCreationTime();
    }

    @Override
    public Watermark getCurrentWatermark() {
        // return the watermark as current time minus the maximum time lag
        return new Watermark(System.currentTimeMillis() - maxTimeLag);
    }
}

public class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks<MyEvent> {

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        return element.getCreationTime();
    }

    @Override
    public Watermark checkAndGetNextWatermark(MyEvent lastElement, long extractedTimestamp) {
        return lastElement.hasWatermarkMarker() ? new Watermark(extractedTimestamp) : null;
    }
}

public class WindowWaterMark {

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub


        String hostName = "localhost";
        Integer port = Integer.parseInt("8001");

        // set up the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(2);
         env.getConfig().setAutoWatermarkInterval(9000);

        // get input data
        DataStream<String> text = env.socketTextStream(hostName, port);

        DataStream<Tuple3<String, Long, Integer>> counts =
    
        text.filter(new FilterClass()).map(new LineSplitter())
        .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple3<String, Long, Integer>>(){

            private long  currentMaxTimestamp = 0l;
            private final long maxOutOfOrderness = 10000l;
            
            @Override
            public long extractTimestamp(Tuple3<String, Long, Integer> element,
                    long previousElementTimestamp) {
                // TODO Auto-generated method stub
                long timestamp= element.f1;
                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                System.out.println("get timestamp is "+timestamp+" currentMaxTimestamp "+currentMaxTimestamp);
                return timestamp;
            }

            @Override
            public Watermark getCurrentWatermark() {
                // TODO Auto-generated method stub
                System.out.println("wall clock is "+ System.currentTimeMillis() +" new watermark "+(currentMaxTimestamp - maxOutOfOrderness));
                   return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                    
            }
            
        })
        .keyBy(0)
        .timeWindow(Time.seconds(20))
//        .allowedLateness(Time.seconds(10))
        .sum(2);

        counts.print();

        // execute program
        env.execute("Java WordCount from SocketTextStream Example");
    }


    public static final class LineSplitter implements
            MapFunction<String, Tuple3<String, Long, Integer>> {

        @Override
        public Tuple3<String, Long, Integer> map(String value) throws Exception {
            // TODO Auto-generated method stub
            String[] tokens = value.toLowerCase().split("\\W+");

            long eventtime = Long.parseLong(tokens[1]);
            
            return new Tuple3<String, Long, Integer>(tokens[0], eventtime, 1);
        }
    }

    private static class MyTimestamp extends
            AscendingTimestampExtractor<Tuple3<String, Long, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public long extractAscendingTimestamp(
                Tuple3<String, Long, Integer> element) {
            // TODO Auto-generated method stub
            return element.f1;
        }

    }
    
    
    public static final class FilterClass implements FilterFunction<String>
    {

        @Override
        public boolean filter(String value) throws Exception {
            // TODO Auto-generated method stub
            if(StringUtils.isNullOrWhitespaceOnly(value))
            {
                return false;
            }
            else
            {
                return true;
            }
        }
        
    }
    
}

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
   return timestamp - (timestamp - offset + windowSize) % windowSize;
}