flink版本: 1.14.6 flink水位生成以及基于水位触发窗口的计算

package flink.shangguigu_test;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
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.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.runtime.operators.util.AssignerWithPeriodicWatermarksAdapter;
import org.apache.flink.streaming.runtime.operators.util.AssignerWithPunctuatedWatermarksAdapter;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Tublingwindow_pv {
    private static final String KAFKA_SERVER = " 10.210.44.10:9092,10.210.44.33:9092,10.210.44.17:9092";
    private static final String KAFKA_TOPIC = "flink_test";
    private static final String KAFKA_GROUP_ID = "flink_test_consumer_group_id_001";


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(1L); //每个一分钟分配一条水位线，默认是200毫秒 设置为1，是为了让每一条数据都生成watermark，便于分析结果数据


        env.enableCheckpointing(5* 1000, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3*1000,5* 1000));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop-ha/user/flink/cluster_yarn/checkpoints");










        DataStream<String> input = env.socketTextStream("127.0.0.1", 8085).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value != null && value != "";
            }
        });

        SingleOutputStreamOperator<String> assign_event_time = input
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<String>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                                    @Override
                                    public long extractTimestamp(String element, long recordTimestamp) {
                                        JSONObject jsonObject = JSON.parseObject(element);
                                        Long operate_time = jsonObject.getLong("operate_time");
                                        return operate_time * 1000L;
                                    }
                                })
                );


        SingleOutputStreamOperator<Tuple2<String,Long>> process = assign_event_time.process(new ProcessFunction<String, Tuple2<String,Long>>() {
            @Override
            public void processElement(String value, Context ctx, Collector<Tuple2<String,Long>> out) throws Exception {

               Long watermark =  ctx.timerService().currentWatermark();
                out.collect(Tuple2.of("时间戳:" + JSON.parseObject(value).getLong("operate_time") +
                        " 事件时间:" + ctx.timestamp() +
                        " 上一条水位:" + ctx.timerService().currentWatermark() +
                        " 当前水位:" + (ctx.timestamp() - 2000 -1 >  watermark ? ctx.timestamp() - 2000 -1 : watermark)
                        ,1L)
                );

            }


        });


        SingleOutputStreamOperator<String> aggregate = process.windowAll(TumblingEventTimeWindows.of(Time.seconds(2))).trigger(EventTimeTrigger.create()).aggregate(new AggregateFunction<Tuple2<String, Long>, Tuple2<StringBuilder,Long>, String>() {
            @Override
            public Tuple2<StringBuilder, Long> createAccumulator() {
                return Tuple2.of(new StringBuilder(),0L);
            }

            @Override
            public Tuple2<StringBuilder, Long> add(Tuple2<String, Long> value, Tuple2<StringBuilder, Long> accumulator) {
                return Tuple2.of(accumulator.f0.append(value.f0).append("-->\n"),value.f1 + accumulator.f1);
            }

            @Override
            public String getResult(Tuple2<StringBuilder, Long> accumulator) {
                return accumulator.f0 + "==" + accumulator.f1;
            }

            @Override
            public Tuple2<StringBuilder, Long> merge(Tuple2<StringBuilder, Long> a, Tuple2<StringBuilder, Long> b) {
                return null;
            }
        });



        aggregate.print();
        env.execute();


    }
}