¡¾½¨ÒéÊղء¿Flink watermark·ÖÎöʵս

env.getConfig().setAutoWatermarkInterval(200)//(ĬÈÏÖµ200ms)
Èç¹ûÒª½ûÓÃwatermark»úÖÆ£¬¿ÉÒÔͨ¹ýÉèÖÃwatermarkÉú³ÉƵÂÊÀ´ÊµÏÖ
env.getConfig().setAutoWatermarkInterval(0)//(ĬÈÏÖµ200ms)

AssignerWithPeriodicWatermarks  £¨ÒѹýÆÚ£©ÖÜÆÚÐÔÉú³Éwatermark
AssignerWithPunctuatedWatermarks£¨ÒѹýÆÚ£© °´ÕÕÖ¸¶¨±ê¼ÇÐÔʼþÉú³Éwatermark

       WatermarkStrategy<Bean> beanWatermarkStrategy = WatermarkStrategy
               .forGenerator(new WatermarkGeneratorSupplier<Bean>() {
                   @Override
                   public WatermarkGenerator<Bean> createWatermarkGenerator(Context context) {
                      return new WatermarkGenerator<Bean>() {

                          /** ×î´óʱ¼ä´Á. */
                           private long maxTimestamp;
                          /** ˮӡÉú³ÉµÄ×î´óÎÞÐò¶È */
                           private final long outOfOrdernessMillis = 0;

                           //watermark±È½ÏÆ÷
                           @Override
                           public void onEvent(Bean event, long eventTimestamp, WatermarkOutput output) {
                               maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
                           }
                           //watermarkÉú³ÉºÍ·¢ËÍ
                           @Override
                           public void onPeriodicEmit(WatermarkOutput output) {
                               output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
                           }
                       };
                   }
               })
//                .noWatermarks()  //´´½¨ÍêÈ«²»Éú³ÉˮӡµÄˮӡ²ßÂÔ¡£ÕâÔÚÖ´Ðд¿´¦Àí»ùÓÚʱ¼äµÄÁ÷´¦ÀíµÄ³¡¾°ÖпÉÄܺÜÓÐÓá£
//                .forMonotonousTimestamps()    //½ô¸ú×î´óʱ¼äʱ¼ä£¬ÍêÈ«²»ÈÝÈÌÂÒÐò
//                .<Bean>forBoundedOutOfOrderness(Duration.ofMillis(0))  //ÔÊÐíÂÒÐòµÄÉú³É²ßÂÔ   ×î´óʱ¼äʱ¼ä-ÈÝ´íʱ¼ä
               .withIdleness(Duration.ofSeconds(5))    //µ±Ä³Ò»²¢·¢³Ù³ÙûÓÐÊý¾Ý½øÀ´Ê±£¬¶à³¤Ê±¼ä·¢ËÍÒ»´ÎwatermarkÖµ
               .withTimestampAssigner(new SerializableTimestampAssigner<Bean>() {
                   @Override
                   public long extractTimestamp(Bean element, long recordTimestamp) {
                       return element.getEventTime();
                   }
               })//watermarkÌáÈ¡²ßÂÔ£¨´ÓÊý¾ÝÖУ©

.<Bean>forBoundedOutOfOrderness(Duration.ofMillis(0))  //ÔÊÐíÂÒÐòµÄÉú³É²ßÂÔ   ×î´óʱ¼äʱ¼ä-ÈÝ´íʱ¼ä

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.api.common.eventtime;

import org.apache.flink.annotation.Public;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * A WatermarkGenerator for situations where records are out of order, but you can place an upper
 * bound on how far the events are out of order. An out-of-order bound B means that once an event
 * with timestamp T was encountered, no events older than {@code T - B} will follow any more.
 *
 * <p>The watermarks are generated periodically. The delay introduced by this watermark strategy is
 * the periodic interval length, plus the out-of-orderness bound.
 */
@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** The maximum timestamp encountered so far. */
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
    }
}

OutputTag<Bean> WMTag = new OutputTag("w_m_tag", TypeInformation.of(Bean.class)) {
        };

        //´°¿Ú°æ
         source.keyBy(Bean::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))

                .allowedLateness(Time.seconds(5))  //´°¿Ú×î´óµÈ´ýʱ¼ä
                
                .sideOutputLateData(WMTag)
                .reduce(new ReduceFunction<Bean>() {
                    @Override
                    public Bean reduce(Bean value1, Bean value2) throws Exception {
                        System.out.println(value2);
                        return value2;
                    }
                })
                ;

 OutputTag<Bean> WMTag = new OutputTag("w_m_tag", TypeInformation.of(Bean.class)) {
        };

        //´°¿Ú°æ
         source.keyBy(Bean::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))

                .allowedLateness(Time.seconds(5))
                
                .sideOutputLateData(WMTag)  //³Ùµ½Êý¾Ý²âÁ÷Êä³ö
                .reduce(new ReduceFunction<Bean>() {
                    @Override
                    public Bean reduce(Bean value1, Bean value2) throws Exception {
                        System.out.println(value2);
                        return value2;
                    }
                })
                ;

.withTimestampAssigner(new SerializableTimestampAssigner<Bean>() {
    @Override
    public long extractTimestamp(Bean element, long recordTimestamp) {
        return element.getEventTime();
    }
})//watermarkÌáÈ¡²ßÂÔ£¨´ÓÊý¾ÝÖУ©

.withIdleness(Duration.ofSeconds(5))

public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> {

    private final WatermarkGenerator<T> watermarks;

    private final IdlenessTimer idlenessTimer;

    /**
        ´´½¨Ò»¸öеÄWatermarksWithIdlenessÉú³ÉÆ÷£¬ÓÃÓÚÔÚ¸ø¶¨³¬Ê±µÄÇé¿öϼì²â¸ø¶¨µÄÉú³ÉÆ÷¿ÕÏÐ״̬¡£
        ²ÎÊý£º
        watermarks¨CԭʼˮӡÉú³ÉÆ÷¡£
        idleTimeout¨C¿ÕÏмì²âµÄ³¬Ê±¡£
     */
    public WatermarksWithIdleness(WatermarkGenerator<T> watermarks, Duration idleTimeout) {
        this(watermarks, idleTimeout, SystemClock.getInstance());
    }

    @VisibleForTesting
    WatermarksWithIdleness(WatermarkGenerator<T> watermarks, Duration idleTimeout, Clock clock) {
        checkNotNull(idleTimeout, "idleTimeout");
        checkArgument(
                !(idleTimeout.isZero() || idleTimeout.isNegative()),
                "idleTimeout must be greater than zero");
        this.watermarks = checkNotNull(watermarks, "watermarks");
        this.idlenessTimer = new IdlenessTimer(clock, idleTimeout);
    }

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        watermarks.onEvent(event, eventTimestamp, output);
        idlenessTimer.activity();
    }
        @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        if (idlenessTimer.checkIfIdle()) {
            output.markIdle();
        } else {
            watermarks.onPeriodicEmit(output);
        }
    }
}

  //½¨±í
        tenv.executeSql( "CREATE TABLE t_kafka("
                +"    shopId string, "
                +"    shopName string,"
                +"    eventTime bigint, "//flink-sqlÖУ¬watermarkµÄÊý¾ÝÀ´Ô´×Ö¶ÎÐèÒªÊÇtimestamp(3)ÀàÐÍ
                +"    rt as to_timestamp_ltz(eventTime,3), "//ËùÒÔÎÒÃÇÏëҪʹÓÃËû,±ØÐëÒª´ÓÊý¾ÝÖиñʽ»¯Ê±¼ä
                +"    orderId string, "
                +"    orderAmount bigint, "
                +"    WATERMARK FOR rt AS rt - INTERVAL '3' SECOND" //watermark_Field-interval ¡®int¡¯ date_tpyeΪboundoutoforderness»úÖÆ
//              +"    et timestamp(3) metadata from 'rowtime'"//Èç¹ûÄãµÄʱ¼ä×ֶΣ¬ÔÚkafkaµÄkeyÖУ¬ÐèÒªÒÔÔªÊý¾ÝµÄÐÎʽ½øÐÐÌáÈ¡£¬¾ßÌ巽ʽ²Î¿¼¹ÙÍøÑùÀý
                +") WITH (                                                   "
                +"  'connector' = 'kafka',                                   "
                +"  'topic' = 'test',                                        "
                +"  'properties.bootstrap.servers' = '192.168.110.128:9092',     "
                +"  'properties.group.id' = 'testGroup',                     "
                +"  'scan.startup.mode' = 'earliest-offset',                 "
                +"  'format' = 'json',                                       "
                +"  'json.fail-on-missing-field' = 'false',                  "
                +"  'json.ignore-parse-errors' = 'true'                      "
                +")                                                          "
        );

//´òÓ¡±í½á¹¹
tenv.executeSql("desc  t_kafka").print();

//²é¿´ÔÚflinksqlÖеÄwatermark±íÏÖЧ¹û
env.executeSql("select eventTime,CURRENT_WATERMARK(rt) as wmTime from t_kafka").print();

//ÑÝʾ¶þ£¬±íתÁ÷ºóµÄwatermark±íÏÖЧ¹û--¿ÉÒԼ̳Ð
        //1.±íתÁ÷ʱ£¬²ÎÊýÐèÒª´«Èë±í¶ø²»ÊÇsql»òÕß±íÃû£¬ËùÒÔÐèÒªÏÈÌáÈ¡³ö±í
        Table t_kafka = tenv.from("t_kafka");
        DataStream<Row> rowDataStream = tenv.toDataStream(t_kafka); //apply-only Á÷
//        tenv.toChangelogStream(t_kafka);//»Ø³·Á÷ +u -u -d

        SingleOutputStreamOperator<Row> process = rowDataStream.process(new ProcessFunction<Row, Row>() {
            @Override
            public void processElement(Row value, ProcessFunction<Row, Row>.Context ctx, Collector<Row> out) throws Exception {
                long l = ctx.timerService().currentWatermark();
//                System.out.println("±íתÁ÷ʱµÄwatermark:" + l);
                out.collect(value);
            }
        });
//        process.print();
        Table table = tenv.fromDataStream(process);  //µ±ÓÉÁ÷ת±íʱ£¬´Ë±í²»»á´ÓÉϲã¼Ì³ÐÈκÎÔªÊý¾ÝÐÅÏ¢£¬°üÀ¨watermark£¬ÐèÒªÖØж¨Òå
        table.printSchema();