flink1.12.x Transformation-keyBy算子

作者：马育民 • • 阅读：10089

为什么要 keyBy 分组

对数据 分组 主要是为了后续的 聚合操作，即：对同组数据进行聚合分析，如：求最大、最小值，求和等

keyBy会将一个 DataStream 转化为一个 KeyedStream，聚合操作会将 KeyedStream 转化为 DataStream。如果聚合前每个元素数据类型是 T，聚合后的数据类型仍为 T。

keyBy 的作用

根据某种属性或数据的某个字段进行分组，对一个分组内的数据进行处理。

如下图：

keyBy 算子根据元素的 形状 对数据进行分组，相同形状的元素被分到了一起，可被后续算子统一处理。

比如：电商用户行为日志把所有用户的行为都记录了下来，如果要分析某一个用户行为，需要先根据用户进行分组。

提示： 没有 groupBy()

执行过程

根据 key 的 hash 值，进行重分区

api

keyBy()方法

通过 source 调用 keyBy() 方法，如下：

public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key)

需要传入 KeySelector 对象

KeySelector

源码如下：

// 通过注解可知，支持lambda表达式
@FunctionalInterface
@Public
public interface KeySelector<IN, KEY> extends Function, Serializable {
    KEY getKey(IN var1) throws Exception;
}

泛型解释：

  • IN：传入的数据类型
  • KEY：key的数据类型

getKey(IN var1) 方法：返回 key

案例

根据用户名分组

主类

package keyBy;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class TestKeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);

        SingleOutputStreamOperator<UserLog> mapSOSO = dss.map(s -> {
            ObjectMapper mapper = new ObjectMapper();
            UserLog result = mapper.readValue(s, UserLog.class);
            return result;
        });

        KeyedStream<UserLog, String> ks = mapSOSO.keyBy(new MyKeySelector());

        ks.print();

        env.execute();
    }

    /**
     * 第一个泛型，表示原数据的类型
     * 第二个泛型，表示key的数据类型
     */
    public static class MyKeySelector implements KeySelector<UserLog, String> {


        @Override
        public String getKey(UserLog userLog) throws Exception {
            return userLog.getName();
        }
    }

}

lambda表达式写法

KeyedStream<UserLog, String> ks = mapSOSO.keyBy(item -> item.getName());

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

创建生产者、指定主题

登录 hadoop1，执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink

然后输入：

{"name":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
{"name":"lili","url":"/dior.html","time":"2020-09-01 10:00:00"}
{"name":"lili","url":"/jianpan.html","time":"2020-09-01 10:00:00"}

执行结果

在 idea 控制台中打印结果中，看不出来特殊的变化。因为只是根据 key 进行分组，还没有进行聚合操作

参考

https://www.jianshu.com/p/5cf06e7e8d71

原文出处：/show_1IX43MKruM0s.html

吉ICP备15003043号