Flink reduce 作用实例

墨蓝 2021-09-24 11:02 442阅读 0赞

**reduce作用：把2个类型相同的值合并成1个，对组内的所有值连续使用reduce，直到留下最后一个值！**

package reduce;
    
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    /**
     * @Author you guess
     * @Date 2020/6/17 20:52
     * @Version 1.0
     * @Desc
     */
    public class DataStreamReduceTest {
    
        public static void main(String[] args) throws Exception {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStreamSource<Tuple3<String, String, Integer>> src1 = env.addSource(new SourceFunction<Tuple3<String, String, Integer>>() {
                @Override
                public void run(SourceContext<Tuple3<String, String, Integer>> ctx) throws Exception {
                    ctx.collect(Tuple3.of("Lisi", "Math", 1));
                    ctx.collect(Tuple3.of("Lisi", "English", 2));
                    ctx.collect(Tuple3.of("Lisi", "Chinese", 3));
    
                    ctx.collect(Tuple3.of("Zhangsan", "Math", 4));
                    ctx.collect(Tuple3.of("Zhangsan", "English", 5));
                    ctx.collect(Tuple3.of("Zhangsan", "Chinese", 6));
                }
    
                @Override
                public void cancel() {
    
                }
            }, "source1");
    
    //        src1.print();
    //        7> (Zhangsan,Chinese,6)
    //        4> (Lisi,Chinese,3)
    //        2> (Lisi,Math,1)
    //        5> (Zhangsan,Math,4)
    //        3> (Lisi,English,2)
    //        6> (Zhangsan,English,5)
    
    
            /**
             * 代码段2
             */
    //        src1.keyBy(0).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 {
    //                return Tuple3.of(value1.f0, "总分:", value1.f2 + value2.f2);
    //            }
    //        }).print();
    //        1> (Lisi,Math,1)
    //        11> (Zhangsan,Math,4)
    //        1> (Lisi,总分:,3)
    //        11> (Zhangsan,总分:,9)
    //        1> (Lisi,总分:,6)
    //        11> (Zhangsan,总分:,15)
    
    
            /**
             * 代码段3，与代码段2 同义
             */
            src1.keyBy(0).reduce((value1, value2) -> Tuple3.of(value1.f0, "总分:", value1.f2 + value2.f2)).print();
    //        1> (Lisi,Math,1)
    //        11> (Zhangsan,Math,4)
    //        1> (Lisi,总分:,3)
    //        11> (Zhangsan,总分:,9)
    //        1> (Lisi,总分:,6)
    //        11> (Zhangsan,总分:,15)
    
            env.execute("Flink DataStreamReduceTest by Java");
        }
    
    
    }

前面几个aggregation是几个较为特殊的操作，对分组数据进行处理更为通用的方法是使用`reduce`算子。

![format_png][]

上图展示了`reduce`算子的原理：**`reduce`在按照同一个Key分组的数据流上生效，它接受两个输入，生成一个输出，即两两合一地进行汇总操作，生成一个同类型的新元素。**

![format_png 1][]

[https://mp.weixin.qq.com/s/2vcKteQIyj31sVrSg1R\_2Q][https_mp.weixin.qq.com_s_2vcKteQIyj31sVrSg1R_2Q]

**DataStreamSource没有aggregate(min minby max maxby sum等)、reduce操作；**

**KeyedStream、AllWindowedStream、DataSet有aggregate(min minby max maxby sum等)、reduce操作；**

[Flink ,Min MinBy Max MaxBy sum实例][Flink _Min MinBy Max MaxBy sum]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 4][]

flink 1.9.2,java1.8

### 源码：注意看注释： ###

/**
     * Base interface for Reduce functions. Reduce functions combine groups of elements to
     * a single value, by taking always two elements and combining them into one. Reduce functions
     * may be used on entire data sets, or on grouped data sets. In the latter case, each group is reduced
     * individually.
     *
     * <p>For a reduce functions that work on an entire group at the same time (such as the
     * MapReduce/Hadoop-style reduce), see {@link GroupReduceFunction}. In the general case,
     * ReduceFunctions are considered faster, because they allow the system to use more efficient
     * execution strategies.
     *
     * <p>The basic syntax for using a grouped ReduceFunction is as follows:
     * <pre>{@code
     * DataSet<X> input = ...;
     *
     * DataSet<X> result = input.groupBy(<key-definition>).reduce(new MyReduceFunction());
     * }</pre>
     *
     * <p>Like all functions, the ReduceFunction needs to be serializable, as defined in {@link java.io.Serializable}.
     *
     * @param <T> Type of the elements that this function processes.
     */
    @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;
    }

DataSet下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 5][]

[format_png]: /images/20210923/5b68d46d4bb24661979312d271991c04.png
[format_png 1]: /images/20210923/e7209114331e4a5f8c34d816d5689c8f.png
[https_mp.weixin.qq.com_s_2vcKteQIyj31sVrSg1R_2Q]: https://mp.weixin.qq.com/s/2vcKteQIyj31sVrSg1R_2Q
[Flink _Min MinBy Max MaxBy sum]: https://mp.csdn.net/console/editor/html/106766341
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70]: /images/20210923/4f7aa92b6d0a48bbb61b33423c45f634.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 1]: /images/20210923/9437b7dd938648518c7485924ead7b91.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 2]: /images/20210923/806f42ee2d2249e291ab26ba83e9c6a5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 3]: /images/20210923/071645841c874497b2e5f2c471dc09bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 4]: /images/20210923/e47475d87eb74e7283033c42421ff328.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAwMDIxODQ_size_16_color_FFFFFF_t_70 5]: /images/20210923/60c630ddd5f4400ca067b141436a5914.png