Flink的CoGroup如何使用

這篇文章主要介紹“Flink的CoGroup如何使用”，在日常操作中，相信很多人在Flink的CoGroup如何使用問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”Flink的CoGroup如何使用”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

CoGroup算子：將兩個(gè)數(shù)據(jù)流按照key進(jìn)行g(shù)roup分組，并將數(shù)據(jù)流按key進(jìn)行分區(qū)的處理，最終合成一個(gè)數(shù)據(jù)流（與join有區(qū)別，不管key有沒(méi)有關(guān)聯(lián)上,最終都會(huì)合并成一個(gè)數(shù)據(jù)流）

示例環(huán)境

java.version: 1.8.x
flink.version: 1.11.1

示例數(shù)據(jù)源 （項(xiàng)目碼云下載）

Flink 系例 之 搭建開(kāi)發(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算子：將兩個(gè)數(shù)據(jù)流按照key進(jìn)行g(shù)roup分組，并將數(shù)據(jù)流按key進(jìn)行分區(qū)的處理，最終合成一個(gè)數(shù)據(jù)流（與join有區(qū)別，不管key有沒(méi)有關(guān)聯(lián)上,最終都會(huì)合并成一個(gè)數(shù)據(jù)流）
 */
public class CoGroup {

    /**
     * 兩個(gè)數(shù)據(jù)流集合，對(duì)相同key進(jìn)行內(nèi)聯(lián)，分配到同一個(gè)窗口下，合并并打印
     * @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 自動(dòng)添加水印調(diào)度時(shí)間
        //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)
                //添加水印窗口,如果不添加，則時(shí)間窗口會(huì)一直等待水印事件時(shí)間，不會(huì)執(zhí)行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)
                //添加水印窗口,如果不添加，則時(shí)間窗口會(huì)一直等待水印事件時(shí)間，不會(huì)執(zhí)行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();
                            }
                        })
                );

        //對(duì)dataStream1和dataStream2兩個(gè)數(shù)據(jù)流進(jìn)行關(guān)聯(lián)，沒(méi)有關(guān)聯(lián)也保留
        //Datastream 3
        DataStream<String> newDataStream = dataStream1.coGroup(dataStream2)
                .where(new KeySelector<Tuple3<String, String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f1;
                    }
                })
                .equalTo(t3->t3.f1)
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .apply(new CoGroupFunction<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple3<String, String, Integer>> first, Iterable<Tuple3<String, String, Integer>> second, Collector<String> out) throws Exception {
                        StringBuilder sb = new StringBuilder();
                        Gson gson = new Gson();
                        //datastream1的數(shù)據(jù)流集合
                        for (Tuple3<String, String, Integer> tuple3 : first) {
                            sb.append(gson.toJson(tuple3)).append("\n");
                        }
                        //datastream2的數(shù)據(jù)流集合
                        for (Tuple3<String, String, Integer> 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é)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！