快捷導(dǎo)航

Flink實(shí)現(xiàn)特定統(tǒng)計(jì)的歸約聚合reduce操作

更新時(shí)間：2023年02月08日 11:55:25 作者：響徹天堂丶

這篇文章主要介紹了Flink實(shí)現(xiàn)特定統(tǒng)計(jì)的歸約聚合reduce操作，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧

如果說(shuō)簡(jiǎn)單聚合是對(duì)一些特定統(tǒng)計(jì)需求的實(shí)現(xiàn)，那么 reduce 算子就是一個(gè)一般化的聚合統(tǒng)計(jì)操作了。從大名鼎鼎的 MapReduce 開(kāi)始，我們對(duì) reduce 操作就不陌生：它可以對(duì)已有的

數(shù)據(jù)進(jìn)行歸約處理，把每一個(gè)新輸入的數(shù)據(jù)和當(dāng)前已經(jīng)歸約出來(lái)的值，再做一個(gè)聚合計(jì)算。與簡(jiǎn)單聚合類(lèi)似，reduce 操作也會(huì)將 KeyedStream 轉(zhuǎn)換為 DataStream。它不會(huì)改變流的元

素?cái)?shù)據(jù)類(lèi)型，所以輸出類(lèi)型和輸入類(lèi)型是一樣的。調(diào)用 KeyedStream 的 reduce 方法時(shí)，需要傳入一個(gè)參數(shù)，實(shí)現(xiàn) ReduceFunction 接口。接口在源碼中的定義如下：

@Public
@FunctionalInterface
public interface ReduceFunction<T> extends Function, Serializable {
    /**
     * The core method of ReduceFunction, combining two values into one value of the same type. The
     * reduce function is consecutively applied to all values of a group until only a single value
     * remains.
     *
     * @param value1 The first value to combine.
     * @param value2 The second value to combine.
     * @return The combined value of both input values.
     * @throws Exception This method may throw exceptions. Throwing an exception will cause the
     *     operation to fail and may trigger recovery.
     */
    T reduce(T value1, T value2) throws Exception;
}

ReduceFunction 接口里需要實(shí)現(xiàn) reduce()方法，這個(gè)方法接收兩個(gè)輸入事件，經(jīng)過(guò)轉(zhuǎn)換處理之后輸出一個(gè)相同類(lèi)型的事件；所以，對(duì)于一組數(shù)據(jù)，我們可以先取兩個(gè)進(jìn)行合并，然后再

將合并的結(jié)果看作一個(gè)數(shù)據(jù)、再跟后面的數(shù)據(jù)合并，最終會(huì)將它“簡(jiǎn)化”成唯一的一個(gè)數(shù)據(jù)，這也就是 reduce“歸約”的含義。在流處理的底層實(shí)現(xiàn)過(guò)程中，實(shí)際上是將中間“合并的結(jié)果”

作為任務(wù)的一個(gè)狀態(tài)保存起來(lái)的；之后每來(lái)一個(gè)新的數(shù)據(jù)，就和之前的聚合狀態(tài)進(jìn)一步做歸約。

其實(shí)，reduce 的語(yǔ)義是針對(duì)列表進(jìn)行規(guī)約操作，運(yùn)算規(guī)則由 ReduceFunction 中的 reduce方法來(lái)定義，而在 ReduceFunction 內(nèi)部會(huì)維護(hù)一個(gè)初始值為空的累加器，注意累加器的類(lèi)型

和輸入元素的類(lèi)型相同，當(dāng)?shù)谝粭l元素到來(lái)時(shí)，累加器的值更新為第一條元素的值，當(dāng)新的元素到來(lái)時(shí)，新元素會(huì)和累加器進(jìn)行累加操作，這里的累加操作就是 reduce 函數(shù)定義的運(yùn)算規(guī)

則。然后將更新以后的累加器的值向下游輸出。

我們可以單獨(dú)定義一個(gè)函數(shù)類(lèi)實(shí)現(xiàn) ReduceFunction 接口，也可以直接傳入一個(gè)匿名類(lèi)。當(dāng)然，同樣也可以通過(guò)傳入 Lambda 表達(dá)式實(shí)現(xiàn)類(lèi)似的功能。與簡(jiǎn)單聚合類(lèi)似，reduce 操作也會(huì)將 KeyedStream 轉(zhuǎn)換為 DataStrema。它不會(huì)改變流的元素?cái)?shù)據(jù)類(lèi)型，所以輸出類(lèi)型和輸入類(lèi)型是一樣的。下面我們來(lái)看一個(gè)稍復(fù)雜的例子。

我們將數(shù)據(jù)流按照用戶(hù) id 進(jìn)行分區(qū)，然后用一個(gè) reduce 算子實(shí)現(xiàn) sum 的功能，統(tǒng)計(jì)每個(gè)用戶(hù)訪問(wèn)的頻次；進(jìn)而將所有統(tǒng)計(jì)結(jié)果分到一組，用另一個(gè) reduce 算子實(shí)現(xiàn) maxBy 的功能，記錄所有用戶(hù)中訪問(wèn)頻次最高的那個(gè)，也就是當(dāng)前訪問(wèn)量最大的用戶(hù)是誰(shuí)。

package com.rosh.flink.test;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
/**
 * 我們將數(shù)據(jù)流按照用戶(hù) id 進(jìn)行分區(qū)，然后用一個(gè) reduce 算子實(shí)現(xiàn) sum 的功能，統(tǒng)計(jì)每個(gè)
 * 用戶(hù)訪問(wèn)的頻次；進(jìn)而將所有統(tǒng)計(jì)結(jié)果分到一組，用另一個(gè) reduce 算子實(shí)現(xiàn) maxBy 的功能，
 * 記錄所有用戶(hù)中訪問(wèn)頻次最高的那個(gè)，也就是當(dāng)前訪問(wèn)量最大的用戶(hù)是誰(shuí)。
 */
public class TransReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //隨機(jī)生成數(shù)據(jù)
        Random random = new Random();
        List<Integer> userIds = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            userIds.add(random.nextInt(5));
        }
        DataStreamSource<Integer> userIdDS = env.fromCollection(userIds);
        //每個(gè)ID訪問(wèn)記錄一次
        SingleOutputStreamOperator<Tuple2<Integer, Long>> mapDS = userIdDS.map(new MapFunction<Integer, Tuple2<Integer, Long>>() {
            @Override
            public Tuple2<Integer, Long> map(Integer value) throws Exception {
                return new Tuple2<>(value, 1L);
            }
        });
        //統(tǒng)計(jì)每個(gè)user訪問(wèn)多少次
        SingleOutputStreamOperator<Tuple2<Integer, Long>> sumDS = mapDS.keyBy(tuple -> tuple.f0).reduce(new ReduceFunction<Tuple2<Integer, Long>>() {
            @Override
            public Tuple2<Integer, Long> reduce(Tuple2<Integer, Long> value1, Tuple2<Integer, Long> value2) throws Exception {
                return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
            }
        });
        sumDS.print("sumDS  ->>>>>>>>>>>>>");
        //把所有分區(qū)合并，求出最大的訪問(wèn)量
        SingleOutputStreamOperator<Tuple2<Integer, Long>> maxDS = sumDS.keyBy(key -> true).reduce(new ReduceFunction<Tuple2<Integer, Long>>() {
            @Override
            public Tuple2<Integer, Long> reduce(Tuple2<Integer, Long> value1, Tuple2<Integer, Long> value2) throws Exception {
                if (value1.f1 > value2.f1) {
                    return value1;
                } else {
                    return value2;
                }
            }
        });
        maxDS.print("maxDS ->>>>>>>>>>>");
        env.execute("TransReduceTest");
    }
}