Flink Window那些事——ReduceFunction窗口函數

ReduceFunction含義
ReduceFunction定義了如何把兩個輸入的元素進行合並來生成相同類型的輸出元素的過程，Flink使用ReduceFunction來對窗口中的元素進行增量聚合

package com.lynch.stream.window; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; /** * 測試ReduceFunction * */
public class TestReduceFunctionOnWindow { public static void main(String[] args) throws Exception{ //獲取執行環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //讀取數據
        DataStream<Tuple3<String,String,Integer>> input = env.fromElements(ENGLISH); //keyBy(0) 計算班級總成績，下標0表示班級 //countWindow(2) 根據元素個數對數據流進行分組切片，達到2個，觸發窗口進行計算
        DataStream<Tuple3<String,String,Integer>>  totalPoints = input.keyBy(0).countWindow(2).reduce(new ReduceFunction<Tuple3<String, String, Integer>>() { @Override public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> value1, Tuple3<String, String, Integer> value2) throws Exception { //效果如下： //(class1,張三,100) //(class1,李四,30) //==============
                System.out.println("" + value1); System.out.println("" + value2); System.out.println("=============="); return new Tuple3<>(value1.f0, value1.f1, value1.f2+value2.f2); } }); //輸出結果 //效果如下： //2> (class1,張三,130)
 totalPoints.print(); env.execute("TestReduceFunctionOnWindow"); } /** * 定義班級的三元數組 */
    public static final Tuple3[] ENGLISH = new Tuple3[]{ //班級 姓名 成績
            Tuple3.of("class1","張三",100), Tuple3.of("class1","李四",30), Tuple3.of("class1","王五",70), Tuple3.of("class2","趙六",50), Tuple3.of("class2","小七",40), Tuple3.of("class2","小八",10), }; }

 

ReduceFunction執行返回結果

(class2,趙六,50) (class2,小七,40) ==============

1> (class2,趙六,90) (class1,張三,100) (class1,李四,30) ==============

2> (class1,張三,130)