flink架构师3-高级开发（window watermark）

一、 window 机制、 0；15~

4、 window 机制介绍 2；15~

4.1 windows类型介绍 2：15~

session 会话窗口简介： 2：18 ~2：20

各类窗口代码演示 2：20~2：29

滑动滚动window操作 2：29~ 2：37

单词计数案例简介(自定义window)： 2：38~ 2:50

实时计算单词出现的次数，但是并不是每次接受到单词以后就输出单词出现的次数，⽽是当过了5秒以后没

收到这个单词，就输出这个单词的次数

解决问题的思路

1. 利⽤state存储key，count和key到达的时间
2. 没接收到⼀个单词，更新状态中的数据
3. 对于每个key都注册⼀个定时器，如果过了5秒没接收到这个key到话，那么就触发这个定时器，这个定时器就判断当前的event time是否等于这个key的最后修改时间+5s，如果等于则输出key以及对应的count

/**
* 5秒没有单词输出，则输出该单词的单词次数
*/
public class KeyedProcessFunctionWordCount
 {
 public static void main(String[] args) throws Exception {
 // 1. 初始化⼀个流执⾏环境
 StreamExecutionEnvironment env =
 StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new
Configuration());
 // 设置每个 operator 的并⾏度
 env.setParallelism(1);
 // socket 数据源不是⼀个可以并⾏的数据源
 DataStreamSource<String> dataStreamSource =
 env.socketTextStream("localhost", 9999);
 // 3. Data Process
 // non keyed stream
 DataStream<Tuple2<String, Integer>> wordOnes =
 dataStreamSource.flatMap(new WordOneFlatMapFunction());
 // 3.2 按照单词进⾏分组, 聚合计算每个单词出现的次数
 // keyed stream
 KeyedStream<Tuple2<String, Integer>, Tuple> wordGroup = wordOnes
 .keyBy(0);
 wordGroup.process(new CountWithTimeoutFunction()).print();
 // 5. 启动并执⾏流程序
 env.execute("Streaming WordCount");
 }

 private static class CountWithTimeoutFunction extends
KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Integer>> 
{
    private ValueState<CountWithTimestamp> state;
   @Override
   public void open(Configuration parameters) throws Exception {
     state = getRuntimeContext().getState(new ValueStateDescriptor<CountWithTimestamp>(
 "myState", CountWithTimestamp.class));
 }

 /**
 * 处理每⼀个接收到的单词(元素)
 * @param element 输⼊元素
 * @param ctx 上下⽂
 * @param out ⽤于输出
 * @throws Exception
 */
 @Override
 public void processElement(Tuple2<String, Integer> element, Context ctx, Collector<Tuple2<String, Integer>> out)  
throws Exception {
 // 拿到当前 key 的对应的状态
 CountWithTimestamp currentState = state.value();
 if (currentState == null) {
 currentState = new CountWithTimestamp();
 currentState.key = element.f0; }
 // 更新这个 key 出现的次数
 currentState.count++;
 // 更新这个 key 到达的时间，最后修改这个状态时间为当前的 Processing Time
 currentState.lastModified =
ctx.timerService().currentProcessingTime();
 // 更新状态
 state.update(currentState);
 // 注册⼀个定时器
 // 注册⼀个以 Processing Time 为准的定时器
 // 定时器触发的时间是当前 key 的最后修改时间加上 5 秒
 ctx.timerService()
 .registerProcessingTimeTimer(currentState.lastModified +
5000);
 }
 /**
 * 定时器需要运⾏的逻辑
 * @param timestamp 定时器触发的时间戳
* @param ctx 上下⽂
 * @param out ⽤于输出
 * @throws Exception
 */
 @Override
 public void onTimer(long timestamp, OnTimerContext ctx,  Collector<Tuple2<String, Integer>> out) throws Exception {
 // 先拿到当前 key 的状态
 CountWithTimestamp curr = state.value();
 // 检查这个 key 是不是 5 秒钟没有接收到数据
 if (timestamp == curr.lastModified + 5000) {
 out.collect(Tuple2.of(curr.key, curr.count));
 state.clear();  } 
 }

 }

 private static class CountWithTimestamp 
{
 public String key;
 public int count;
 public long lastModified;
 }

 private static class WordOneFlatMapFunction  implements FlatMapFunction<String, Tuple2<String, Integer>>
 {

 @Override
 public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws
Exception {

 String[] words = line.toLowerCase().split(" ");
 for (String word : words) {
 Tuple2<String, Integer> wordOne = new Tuple2<>(word, 1);
 // 将单词计数 1 的⼆元组输出
 out.collect(wordOne);
 }

 }

 }

}
           

单词计数案例简介(session window)： 2:50~2:51

一行代码：

stream.keyBy(0)

//3: 会话窗⼝ 5s

.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))

.sum(1)

.print();

global 窗口简介(global window)： 2:51~

global window + trigger ⼀起配合才能使⽤

需求：单词每出现三次统计⼀次

stream.keyBy(0)

.window(GlobalWindows.create())

//如果不加这个程序是启动不起来的

.trigger(CountTrigger.of(3))

.sum(1)

.print();

env.execute("SessionWindowTest");

trigger 简介： 2:56~ 3：08

evictor简介：3：08~3：18

window 聚合和join：3：18~ 3：30