Flink的CoGroup如何使用
这篇文章主要介绍“Flink的CoGroup如何使用”,在日常操作中,相信很多人在Flink的CoGroup如何使用问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Flink的CoGroup如何使用”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
在网站制作、成都网站制作中从网站色彩、结构布局、栏目设置、关键词群组等细微处着手,突出企业的产品/服务/品牌,帮助企业锁定精准用户,提高在线咨询和转化,使成都网站营销成为有效果、有回报的无锡营销推广。成都创新互联公司专业成都网站建设10年了,客户满意度97.8%,欢迎成都创新互联客户联系。
CoGroup算子:将两个数据流按照key进行group分组,并将数据流按key进行分区的处理,最终合成一个数据流(与join有区别,不管key有没有关联上,最终都会合并成一个数据流)
示例环境
java.version: 1.8.x flink.version: 1.11.1
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
CoGroup.java
package com.flink.examples.functions; import com.flink.examples.DataSource; import com.google.gson.Gson; import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.CoGroupFunction; 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.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; import java.time.Duration; import java.util.Arrays; import java.util.List; /** * @Description CoGroup算子:将两个数据流按照key进行group分组,并将数据流按key进行分区的处理,最终合成一个数据流(与join有区别,不管key有没有关联上,最终都会合并成一个数据流) */ public class CoGroup { /** * 两个数据流集合,对相同key进行内联,分配到同一个窗口下,合并并打印 * @param args * @throws Exception */ public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //watermark 自动添加水印调度时间 //env.getConfig().setAutoWatermarkInterval(200); List> tuple3List1 = DataSource.getTuple3ToList(); List > tuple3List2 = Arrays.asList( new Tuple3<>("伍七", "girl", 18), new Tuple3<>("吴八", "man", 30) ); //Datastream 1 DataStream > dataStream1 = env.fromCollection(tuple3List1) //添加水印窗口,如果不添加,则时间窗口会一直等待水印事件时间,不会执行apply .assignTimestampsAndWatermarks(WatermarkStrategy . >forBoundedOutOfOrderness(Duration.ofSeconds(2)) .withTimestampAssigner((element, timestamp) -> System.currentTimeMillis())); //Datastream 2 DataStream > dataStream2 = env.fromCollection(tuple3List2) //添加水印窗口,如果不添加,则时间窗口会一直等待水印事件时间,不会执行apply .assignTimestampsAndWatermarks(WatermarkStrategy . >forBoundedOutOfOrderness(Duration.ofSeconds(2)) .withTimestampAssigner(new SerializableTimestampAssigner >() { @Override public long extractTimestamp(Tuple3 element, long timestamp) { return System.currentTimeMillis(); } }) ); //对dataStream1和dataStream2两个数据流进行关联,没有关联也保留 //Datastream 3 DataStream newDataStream = dataStream1.coGroup(dataStream2) .where(new KeySelector , String>() { @Override public String getKey(Tuple3 value) throws Exception { return value.f1; } }) .equalTo(t3->t3.f1) .window(TumblingEventTimeWindows.of(Time.seconds(1))) .apply(new CoGroupFunction , Tuple3 , String>() { @Override public void coGroup(Iterable > first, Iterable > second, Collector out) throws Exception { StringBuilder sb = new StringBuilder(); Gson gson = new Gson(); //datastream1的数据流集合 for (Tuple3 tuple3 : first) { sb.append(gson.toJson(tuple3)).append("\n"); } //datastream2的数据流集合 for (Tuple3 tuple3 : second) { sb.append(gson.toJson(tuple3)).append("\n"); } out.collect(sb.toString()); } }); newDataStream.print(); env.execute("flink CoGroup job"); } }
打印结果
{"f0":"张三","f1":"man","f2":20} {"f0":"王五","f1":"man","f2":29} {"f0":"吴八","f1":"man","f2":30} {"f0":"吴八","f1":"man","f2":30} {"f0":"李四","f1":"girl","f2":24} {"f0":"刘六","f1":"girl","f2":32} {"f0":"伍七","f1":"girl","f2":18} {"f0":"伍七","f1":"girl","f2":18}
到此,关于“Flink的CoGroup如何使用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!
新闻名称:Flink的CoGroup如何使用
文章URL:http://scyanting.com/article/jgcepj.html