Flink操练（四十一）之周期水位线窗口

　　关键代码env。getConfig（）。setAutoWatermarkInterval（601000L）；完整代码packageday04；importorg。apache。flink。api。common。eventtime。SerializableTimestampAssigner；importorg。apache。flink。api。common。eventtime。WatermarkStrategy；importorg。apache。flink。api。common。functions。MapFunction；importorg。apache。flink。api。java。tuple。Tuple2；importorg。apache。flink。streaming。api。datastream。DataStream；importorg。apache。flink。streaming。api。datastream。DataStreamSource；importorg。apache。flink。streaming。api。environment。StreamExecutionEnvironment；importorg。apache。flink。streaming。api。functions。windowing。ProcessWindowFunction；importorg。apache。flink。streaming。api。windowing。assigners。TumblingEventTimeWindows；importorg。apache。flink。streaming。api。windowing。time。Time；importorg。apache。flink。streaming。api。windowing。windows。TimeWindow；importorg。apache。flink。util。Collector；importjava。sql。Timestamp；importjava。time。Duration；program：bigDatalearndescription：水位线测试author：Mr。逗create：2021092409：30publicclassWaterTest2｛publicstaticvoidmain（String〔〕args）｛StreamExecutionEnvironmentenvStreamExecutionEnvironment。getExecutionEnvironment（）；env。setParallelism（1）；每隔1分钟插入一次水位线env。getConfig（）。setAutoWatermarkInterval（601000L）；DataStreamSourceStringsourceenv。socketTextStream（172。17。0。50，9999）；DataStreamTuple2String，Longmapsource。map（newMapFunctionString，Tuple2String，Long（）｛OverridepublicTuple2String，Longmap（Stringv）throwsException｛returnTuple2。of（v。split（）〔0〕，Long。parseLong（v。split（）〔1〕）1000L）；｝｝）；默认每隔200ms的机器时间，插入一次水位线DataStreamTuple2String，Longwatermarksmap。assignTimestampsAndWatermarks（WatermarkStrategy。Tuple2String，LongforBoundedOutOfOrderness（Duration。ofSeconds（0））。withTimestampAssigner（newSerializableTimestampAssignerTuple2String，Long（）｛OverridepubliclongextractTimestamp（Tuple2String，Longv，longl）｛returnv。f1；告诉flink事件时间是哪一个字段｝｝））；DataStreamStringprocesswatermarks。keyBy（vv。f0）5秒的事件时间滚动窗口。window（TumblingEventTimeWindows。of（Time。seconds（5）））。process（newProcessWindowFunctionTuple2String，Long，String，String，TimeWindow（）｛Overridepublicvoidprocess（Stringkey，Contextctx，IterableTuple2String，Longit，CollectorStringout）throwsException｛longstartctx。window（）。getStart（）；longendctx。window（）。getEnd（）；longcountit。spliterator（）。getExactSizeIfKnown（）；迭代器里面共有多少元素out。collect（用户：key在窗口newTimestamp（start）newTimestamp（end）中的pv次数是：count）；｝｝）；process。print（）；StringnameWaterTest2。class。getName（）；try｛env。execute（name）；｝catch（Exceptione）｛e。printStackTrace（）；｝｝｝