正文
Flink入门 - CoGroup和Join
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
/*
*CoGroup
*/
final StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<String, String>> dataStream1 = streamExecutionEnvironment.socketTextStream("127.0.0.1", 9000)
.map(new MapFunction<String, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> map(String s) throws Exception {
List<String> strs = Arrays.asList(s.split(" "));
return new Tuple2<>(strs.get(0), strs.get(1));
}
});
DataStream<Tuple2<String, String>> dataStream2 = streamExecutionEnvironment.socketTextStream("127.0.0.1", 9001)
.map(new MapFunction<String, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> map(String s) throws Exception {
List<String> strs = Arrays.asList(s.split(" "));
return new Tuple2<>(strs.get(0), strs.get(1));
}
});
dataStream1.coGroup(dataStream2)
.where(new KeySelector<Tuple2<String, String>, String>() {
@Override
public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
return stringStringTuple2.f0;
}
})
.equalTo(new KeySelector<Tuple2<String, String>, String>() {
@Override
public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
return stringStringTuple2.f0;
}
})
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
.trigger(CountTrigger.of(1))
.apply(new CoGroupFunction<Tuple2<String, String>, Tuple2<String, String>, String>() {
@Override
public void coGroup(Iterable<Tuple2<String, String>> first, Iterable<Tuple2<String, String>> second, Collector<String> out) throws Exception {
StringBuilder stringBuilder = new StringBuilder("Data Stream1: \n");
first.forEach(item -> stringBuilder.append(item.f0 + "<=>" + item.f1 + "\n"));
stringBuilder.append("Data Stream2: \n");
second.forEach(item -> stringBuilder.append(item.f0 + "<=>" + item.f1 + "\n"));
out.collect(stringBuilder.toString());
}
}).print();
streamExecutionEnvironment.execute();
/**
* Join
* Join条件为两个流中的数据((String, String))的第一个元素相同
*/
final StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<String, String>> dataStream1 = streamExecutionEnvironment.fromElements(new Tuple2<>("hello", "flink"));
DataStream<Tuple2<String, String>> dataStream2 = streamExecutionEnvironment.fromElements(new Tuple2<>("hello", "blink"));
dataStream1.join(dataStream2)
.where(new KeySelector<Tuple2<String, String>, String>() {
@Override
public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
return stringStringTuple2.f0;
}
})
.equalTo(new KeySelector<Tuple2<String, String>, String>() {
@Override
public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
return stringStringTuple2.f0;
}
})
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
.trigger(CountTrigger.of(1))
.apply(new JoinFunction<Tuple2<String, String>, Tuple2<String, String>, String>() {
@Override
public String join(Tuple2<String, String> first, Tuple2<String, String> second) throws Exception {
return first.f1 + "<=>" + second.f1;
}
}).print();
streamExecutionEnvironment.execute();
// 运行结果
2> flink<=>blink
