这篇文章主要讲解了“Flink Join怎么使用”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Flink Join怎么使用”吧!
Join算子:两个数据流通过内部相同的key分区,将窗口内两个数据流相同key数据元素计算后,合并输出(类似于mysql表的inner join操作)
示例环境
java.version: 1.8.x
flink.version: 1.11.1
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
Join.java
package com.flink.examples.functions;
import com.flink.examples.DataSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatJoinFunction;
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 Join算子:两个数据流通过内部相同的key分区,将窗口内两个数据流相同key数据元素计算后,合并输出(类似于mysql表的inner join操作)
*/
public class Join {
/**
* Flink支持了两种Join:Window Join(窗口连接)和Interval Join(时间间隔连接),本示例演示的为Window Join
* 官方文档:https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/stream/operators/joining.html
*/
/**
* 两个数据流集合,对相同key进行内联,分配到同一个窗口下,合并并打印
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// //watermark 自动添加水印调度时间
// env.getConfig().setAutoWatermarkInterval(200);
List<Tuple3<String, String, Integer>> tuple3List1 = DataSource.getTuple3ToList();
List<Tuple3<String, String, Integer>> tuple3List2 = Arrays.asList(
new Tuple3<>("伍七", "girl", 18),
new Tuple3<>("吴八", "man", 30)
);
//Datastream 1
DataStream<Tuple3<String, String, Integer>> dataStream1 = env.fromCollection(tuple3List1)
//添加水印窗口,如果不添加,则时间窗口会一直等待水印事件时间,不会执行apply
.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner((element, timestamp)->System.currentTimeMillis()));
//Datastream 2
DataStream<Tuple3<String, String, Integer>> dataStream2 = env.fromCollection(tuple3List2)
//添加水印窗口,如果不添加,则时间窗口会一直等待水印事件时间,不会执行apply
.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Integer>>() {
@Override
public long extractTimestamp(Tuple3<String, String, Integer> element, long timestamp) {
return System.currentTimeMillis();
}
}));
//Datastream 3
DataStream<String> newDataStream = dataStream1.join(dataStream2)
.where(new KeySelector<Tuple3<String, String, Integer>, String>() {
@Override
public String getKey(Tuple3<String, String, Integer> value) throws Exception {
System.out.println("first name:" + value.f0 + ",sex:" + value.f1);
return value.f1;
}
})
.equalTo(new KeySelector<Tuple3<String, String, Integer>, String>() {
@Override
public String getKey(Tuple3<String, String, Integer> value) throws Exception {
System.out.println("second name:" + value.f0 + ",sex:" + value.f1);
return value.f1;
}
})
.window(TumblingEventTimeWindows.of(Time.seconds(1))
.apply(new FlatJoinFunction<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>, String>() {
@Override
public void join(Tuple3<String, String, Integer> first, Tuple3<String, String, Integer> second, Collector<String> out) throws Exception {
out.collect(first.f0 + "|" + first.f1 + "|" + first.f2 + "|" + second.f0 + "|" + second.f1 + "|" + second.f2);
}
})
;
newDataStream.print();
env.execute("flink Join job");
}
}
打印结果
4> 李四|girl|24|伍七|girl|18
4> 刘六|girl|32|伍七|girl|18
4> 伍七|girl|18|伍七|girl|18
2> 张三|man|20|吴八|man|30
2> 王五|man|29|吴八|man|30
2> 吴八|man|30|吴八|man|30
感谢各位的阅读,以上就是“Flink Join怎么使用”的内容了,经过本文的学习后,相信大家对Flink Join怎么使用这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/u/437309/blog/4672885