点一下关注吧！！！非常感谢！！持续更新！！！
目前已经更新到了：
Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（正在更新！）
章节内容
上节我们完成了如下的内容：
Flink Window 背景总览
Flink Window 滚动时间窗口
基于时间驱动
基于事件驱动
滑动时间窗口
滑动窗口是固定窗口更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。Flink 的滑动时间窗口（Sliding Window）是一种常用的窗口机制，适用于处理流式数据时需要在时间范围内定期计算的场景。滑动窗口会按照指定的窗口大小（window size）和滑动步长（slide interval）不断地划分数据，并对每个窗口内的数据进行聚合计算。
类型特点
窗口长度固定，可以有重叠。
滑动窗口会有重叠部分，因此每个事件可能会被包含在多个窗口中。
滑动窗口更适合定期计算某个时间范围内的聚合值，像是移动平均值、最近一段时间的活跃用户等场景。
关键参数
窗口大小（window size）：每个窗口包含的时间范围，例如 10 秒。
滑动步长（slide interval）：窗口每次滑动的时间步长，例如 5 秒。这意味着每隔 5 秒就会创建一个新的窗口，每个窗口覆盖的时间范围是 10 秒。
基于时间驱动
场景：我们可以每30秒计算一次最近一分钟用户购买的商品数
package icu.wzk; import org.apache.commons.math3.analysis.function.Sin; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple1; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.WindowedStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import java.text.SimpleDateFormat; import java.util.Random; public class SlidingWindow { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999); SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); long timeMillis = System.currentTimeMillis(); int random = new Random().nextInt(10); System.out.println("value: " + value + ", random: " + random + ", timestamp: " + format.format(timeMillis)); return Tuple2.of(value, random); } }); KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream .keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() { @Override public Tuple getKey(Tuple2<String, Integer> value) throws Exception { return Tuple1.of(value.f0); } }); WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> timeWindow = keyedStream .timeWindow(Time.seconds(10), Time.seconds(5)); timeWindow.apply(new MyTimeWindowFunction()).print(); env.execute("SlidingWindow"); } }
基于事件驱动
package icu.wzk; import org.apache.commons.math3.analysis.function.Sin; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple1; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.WindowedStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import java.text.SimpleDateFormat; import java.util.Random; public class SlidingWindow { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999); SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); long timeMillis = System.currentTimeMillis(); int random = new Random().nextInt(10); System.out.println("value: " + value + ", random: " + random + ", timestamp: " + format.format(timeMillis)); return Tuple2.of(value, random); } }); KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream .keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() { @Override public Tuple getKey(Tuple2<String, Integer> value) throws Exception { return Tuple1.of(value.f0); } }); WindowedStream<Tuple2<String, Integer>, Tuple, GlobalWindow> globalWindow = keyedStream .countWindow(3, 2); globalWindow.apply(new MyCountWindowFuntion()).print(); env.execute("SlidingWindow"); } }
会话窗口
由一系列事件组合一个指定时间长度timeout间隙组成，类似于Web应用的Session，也就是一段时间没有接收到新数据会生成新的窗口。
Session窗口分配器通过Session活动来对元素进行分组，Session窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况。
Session窗口在一个固定的时间周期内不再收到元素，即非活动间隔产生，那么这个窗口就会关闭。
一个Session窗口通过一个Session间隔来配置，这个Session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的Session将关闭并且后续的元素将被分配到新的Session窗口去。
类型特点
会话窗口不重叠，没有固定的开始和结束时间
于翻滚窗口和滑动窗口相反，当会话窗口在一段时间内没有接收到元素时会关闭会话窗口。
后续的元素将会被分配到新的会话窗口
基于时间驱动
package icu.wzk; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple1; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.WindowedStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; public class SessionWindow { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999); SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = dataStreamSource .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return null; } }); KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream .keyBy(new KeySelector<Tuple2<String, Integer>, Tuple>() { @Override public Tuple getKey(Tuple2<String, Integer> value) throws Exception { return Tuple1.of(value.f0); } }); WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> window = keyedStream .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10))); window.apply(new MyTimeWindowFunction()).print(); env.execute("SessionWindow"); } }
