Flink 1.17教程：聚合算子（Aggregation）之按键分区（keyBy）

我就是我 2023-10-15 18:20 11阅读 0赞

### 聚合算子（Aggregation） ###

计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并——这就是所谓的“聚合”（`Aggregation`），类似于MapReduce中的reduce操作。

#### 按键分区（keyBy） ####

对于Flink而言，DataStream是没有直接进行聚合的API的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在Flink中，要做聚合，需要先进行分区；这个操作就是通过keyBy来完成的。  
keyBy是聚合前必须要用到的一个算子。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务。  
基于不同的key，流中的数据将被分配到不同的分区中去；这样一来，所有具有相同的key的数据，都将被发往同一个分区。![在这里插入图片描述][f71da6edb499480cacff6ce674901724.png]

在内部，是通过计算key的哈希值（hash code），对分区数进行取模运算来实现的。所以这里key如果是POJO的话，必须要重写hashCode()方法。  
keyBy()方法需要传入一个参数，这个参数指定了一个或一组key。有很多不同的方法来指定key：比如对于Tuple数据类型，可以指定字段的位置或者多个位置的组合；对于POJO类型，可以指定字段的名称（String）；另外，还可以传入Lambda表达式或者实现一个键选择器（KeySelector），用于说明从数据中提取key的逻辑。  
我们可以以id作为key做一个分区操作，代码实现如下：

package com.atguigu.zxl_test;
    
    import com.atguigu.bean.WaterSensor;
    import org.apache.flink.api.java.functions.KeySelector;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class TransKeyBy {
        
        public static void main(String[] args) throws Exception {
        
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 2L, 2),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
            );
    
            // 方式一：使用Lambda表达式
            KeyedStream<WaterSensor, String> keyedStream = stream.keyBy(e -> e.id);
    
            // 添加操作符，例如打印结果  解决报错:No operators defined in streaming topology. Cannot execute.
            keyedStream.print();
    
            // 方式二：使用匿名类实现KeySelector
            /*KeyedStream<WaterSensor, String> keyedStream1 = stream.keyBy(new KeySelector<WaterSensor, String>() {
                @Override
                public String getKey(WaterSensor e) throws Exception {
                    return e.id;
                }
            });
    
            // 添加操作符，例如打印结果 解决报错:No operators defined in streaming topology. Cannot execute.
            keyedStream1.print();*/
    
            env.execute();
        }
    }

执行结果：  
![在这里插入图片描述][f629f4cb9d8e42cd80cbe0027f9faa43.png]

需要注意的是，keyBy得到的结果将不再是DataStream，而是会将DataStream转换为KeyedStream。KeyedStream可以认为是“分区流”或者“键控流”，它是对DataStream按照key的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定key的类型。  
KeyedStream也继承自DataStream，所以基于它的操作也都归属于DataStream API。但它跟之前的转换操作得到的SingleOutputStreamOperator不同，只是一个流的分区操作，并不是一个转换算子。KeyedStream是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如sum，reduce）。

[f71da6edb499480cacff6ce674901724.png]: https://img-blog.csdnimg.cn/f71da6edb499480cacff6ce674901724.png
[f629f4cb9d8e42cd80cbe0027f9faa43.png]: https://img-blog.csdnimg.cn/f629f4cb9d8e42cd80cbe0027f9faa43.png