建站
咨詢
售后
建站咨詢
新聞中心
這篇文章主要介紹“Flink的CoGroup如何使用”，在日常操作中，相信很多人在Flink的CoGroup如何使用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Flink的CoGroup如何使用”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！
在網(wǎng)站制作、成都網(wǎng)站制作中從網(wǎng)站色彩、結(jié)構(gòu)布局、欄目設(shè)置、關(guān)鍵詞群組等細微處著手，突出企業(yè)的產(chǎn)品/服務(wù)/品牌，幫助企業(yè)鎖定精準用戶，提高在線咨詢和轉(zhuǎn)化，使成都網(wǎng)站營銷成為有效果、有回報的無錫營銷推廣。成都創(chuàng)新互聯(lián)公司專業(yè)成都網(wǎng)站建設(shè)10年了，客戶滿意度97.8%，歡迎成都創(chuàng)新互聯(lián)客戶聯(lián)系。
CoGroup算子：將兩個數(shù)據(jù)流按照key進行g(shù)roup分組，并將數(shù)據(jù)流按key進行分區(qū)的處理，最終合成一個數(shù)據(jù)流（與join有區(qū)別，不管key有沒有關(guān)聯(lián)上,最終都會合并成一個數(shù)據(jù)流）
示例環(huán)境
java.version: 1.8.x flink.version: 1.11.1
示例數(shù)據(jù)源 （項目碼云下載）
Flink 系例 之 搭建開發(fā)環(huán)境與數(shù)據(jù)
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算子：將兩個數(shù)據(jù)流按照key進行g(shù)roup分組，并將數(shù)據(jù)流按key進行分區(qū)的處理，最終合成一個數(shù)據(jù)流（與join有區(qū)別，不管key有沒有關(guān)聯(lián)上,最終都會合并成一個數(shù)據(jù)流）
*/
public class CoGroup {
/**
* 兩個數(shù)據(jù)流集合，對相同key進行內(nèi)聯(lián)，分配到同一個窗口下，合并并打印
* @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 自動添加水印調(diào)度時間
//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)
//添加水印窗口,如果不添加，則時間窗口會一直等待水印事件時間，不會執(zhí)行apply
.assignTimestampsAndWatermarks(WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner((element, timestamp) -> System.currentTimeMillis()));
//Datastream 2
DataStream> dataStream2 = env.fromCollection(tuple3List2)
//添加水印窗口,如果不添加，則時間窗口會一直等待水印事件時間，不會執(zhí)行apply
.assignTimestampsAndWatermarks(WatermarkStrategy
.>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner>() {
@Override
public long extractTimestamp(Tuple3 element, long timestamp) {
return System.currentTimeMillis();
}
})
);
//對dataStream1和dataStream2兩個數(shù)據(jù)流進行關(guān)聯(lián)，沒有關(guān)聯(lián)也保留
//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的數(shù)據(jù)流集合
for (Tuple3 tuple3 : first) {
sb.append(gson.toJson(tuple3)).append("\n");
}
//datastream2的數(shù)據(jù)流集合
for (Tuple3 tuple3 : second) {
sb.append(gson.toJson(tuple3)).append("\n");
}
out.collect(sb.toString());
}
});
newDataStream.print();
env.execute("flink CoGroup job");
}
} 打印結(jié)果
{"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}到此，關(guān)于“Flink的CoGroup如何使用”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
標題名稱：Flink的CoGroup如何使用
標題來源：http://fisionsoft.com.cn/article/jgcepj.html
