大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章主要讲解了“Flink的SessionWindow怎么用”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Flink的SessionWindow怎么用”吧!
成都创新互联公司是一家专注于成都网站制作、网站建设、外贸网站建设与策划设计,利辛网站建设哪家好?成都创新互联公司做网站,专注于网站建设10多年,网设计领域的专业建站公司;建站业务涵盖:利辛等地区。利辛做网站价格咨询:18980820575
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> inStream = env.addSource(new MyRichSourceFunction()); DataStream > dataStream = inStream.keyBy((KeySelector , String>) k ->k.f1) //按会话窗口滚动,当2秒之内没有指定分区数据流,则计算一次 //会话窗口是根据在指定时间之后没有活跃的数据接入,则认为窗口结束,进行窗口计算 .window(EventTimeSessionWindows.withGap(Time.seconds(2))) .reduce(new ReduceFunction >() { @Override public Tuple3 reduce(Tuple3 t1, Tuple3 t2) throws Exception { //返回年龄最大的 return t1.f2 > t2.f2 ? t1: t2; } }); dataStream.print(); env.execute("flink EventTimeSessionWindows job"); } /** * 模拟数据持续输出 */ public static class MyRichSourceFunction extends RichParallelSourceFunction > { @Override public void run(SourceContext > ctx) throws Exception { List > 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怎么用这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是创新互联,小编将为大家推送更多相关知识点的文章,欢迎关注!