一、 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秒以后没
收到这个单词，就输出这个单词的次数
解决问题的思路
- 利⽤state存储key，count和key到达的时间
- 没接收到⼀个单词，更新状态中的数据
- 对于每个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");
