这篇文章主要讲解了“Flink的SessionWindow怎么用”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Flink的SessionWindow怎么用”吧!
sessionWindows会话窗口:按不活跃时间切成不同分区窗口,并进行窗口计算
示例环境
java.version: 1.8.xflink.version: 1.11.1
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
SessionWindow.java
import com.flink.examples.DataSource;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.util.List;
/**
* @Description sessionWindows会话窗口:按不活跃时间切成不同分区窗口,并进行窗口计算
*/
public class SessionWindow {
/**
* 遍历集合,返回会话滑动窗口下按不活跃时间切分后的,每个窗口下性别分区里最大年龄数据记录
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//设置流处理时间事件,对于会话窗口必需设置此时间类型,有三种类型:
//1.ProcessingTime:以operator处理的时间为准,它使用的是机器的系统时间来作为data stream的时间
//2.IngestionTime:以数据进入flink streaming data flow的时间为准
//3.EventTime:以数据自带的时间戳字段为准,应用程序需要指定如何从record中抽取时间戳字段
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
env.setParallelism(4);
DataStream<Tuple3<String, String, Integer>> inStream = env.addSource(new MyRichSourceFunction());
DataStream<Tuple3<String, String, Integer>> dataStream = inStream.keyBy((KeySelector<Tuple3<String, String, Integer>, String>) k ->k.f1)
//按会话窗口滚动,当2秒之内没有指定分区数据流,则计算一次
//会话窗口是根据在指定时间之后没有活跃的数据接入,则认为窗口结束,进行窗口计算
.window(EventTimeSessionWindows.withGap(Time.seconds(2)))
.reduce(new ReduceFunction<Tuple3<String, String, Integer>>() {
@Override
public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> t1, Tuple3<String, String, Integer> t2) throws Exception {
//返回年龄最大的
return t1.f2 > t2.f2 ? t1: t2;
}
});
dataStream.print();
env.execute("flink EventTimeSessionWindows job");
}
/**
* 模拟数据持续输出
*/
public static class MyRichSourceFunction extends RichParallelSourceFunction<Tuple3<String, String, Integer>> {
@Override
public void run(SourceContext<Tuple3<String, String, Integer>> ctx) throws Exception {
List<Tuple3<String, String, Integer>> tuple3List = DataSource.getTuple3ToList();
for (Tuple3 tuple3 : tuple3List){
ctx.collect(tuple3);
//1秒钟输出一个
Thread.sleep(2 * 1000);
}
}
@Override
public void cancel() {
try{
super.close();
}catch (Exception e){
e.printStackTrace();
}
}
}
}
打印结果
2> (张三,man,20)
4> (李四,girl,24)
2> (王五,man,29)
4> (刘六,girl,32)
2> (吴八,man,30)
感谢各位的阅读,以上就是“Flink的SessionWindow怎么用”的内容了,经过本文的学习后,相信大家对Flink的SessionWindow怎么用这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/u/437309/blog/4889773