Flink 1.17教程：转换算子（Transformation）之归约聚合算子（reduce）

素颜马尾好姑娘i 2023-10-15 18:18 18阅读 0赞

#### 归约聚合（reduce） ####

reduce可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。  
reduce操作也会将KeyedStream转换为DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。  
调用KeyedStream的reduce方法时，需要传入一个参数，实现ReduceFunction接口。接口在源码中的定义如下：

public interface ReduceFunction<T> extends Function, Serializable {
        
        T reduce(T value1, T value2) throws Exception;
    }

ReduceFunction接口里需要实现reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件。在流处理的底层实现过程中，实际上是将中间“合并的结果”作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。  
我们可以单独定义一个函数类实现ReduceFunction接口，也可以直接传入一个匿名类。当然，同样也可以通过传入Lambda表达式实现类似的功能。  
为了方便后续使用，定义一个WaterSensorMapFunction：

public class WaterSensorMapFunction implements MapFunction<String,WaterSensor> {
        
        @Override
        public WaterSensor map(String value) throws Exception {
        
            String[] datas = value.split(",");
            return new WaterSensor(datas[0],Long.valueOf(datas[1]) ,Integer.valueOf(datas[2]) );
        }
    }

> 案例：使用reduce实现max和maxBy的功能。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    env
       .socketTextStream("hadoop102", 7777)
       .map(new WaterSensorMapFunction())
       .keyBy(WaterSensor::getId)
       .reduce(new ReduceFunction<WaterSensor>()
       {
        
           @Override
           public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
        
               System.out.println("Demo7_Reduce.reduce");
    
               int maxVc = Math.max(value1.getVc(), value2.getVc());
               //实现max(vc)的效果  取最大值，其他字段以当前组的第一个为主
               //value1.setVc(maxVc);
               //实现maxBy(vc)的效果  取当前最大值的所有字段
               if (value1.getVc() > value2.getVc()){
        
                   value1.setVc(maxVc);
                   return value1;
               }else {
        
                   value2.setVc(maxVc);
                   return value2;
               }
           }
       })
       .print();
    env.execute();

reduce同简单聚合算子一样，也要针对每一个key保存状态。因为状态不会清空，所以我们需要将reduce算子作用在一个有限key的流上。