说明

在 flink1.12.x Transformation-keyBy算子 中，执行 keyBy 分组后，接下来开始进行聚合运算

本文介绍 max() 方法的特点

案例

根据用户名分组，然后取出最近的时间，使用 max() 方法

主类

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.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 求最大值2 {
    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;
        });

        SingleOutputStreamOperator<UserLog> maxSOSO = mapSOSO.keyBy(item -> item.getName()).max("time");

        maxSOSO.print();

        env.execute();
    }


}

启动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

第一次输入

在kafka控制输入下面内容：

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

执行结果

在 idea 控制台中打印结果如下：

1> UserLog{name='lili', url='/dior.html', time=Tue Sep 01 12:00:00 CST 2020}

解释：因为只有一条数据，所以就输出该条数据

第二次输入

在kafka控制输入下面内容：

{"name":"lili","url":"/lv.html","time":"2020-09-01 13:00:00"}

注意：用户名也是：lili，但时间是 2020-09-01 13:00:00

执行结果

在 idea 控制台中打印结果如下：

1> UserLog{name='lili', url='/dior.html', time=Tue Sep 01 13:00:00 CST 2020}

解释：输出结果中：

• url 是第一个数据的，
• time 是第二个数据的，因为第二个数据的time更晚

关键：这是 max() 的特点

第三次输入

在kafka控制输入下面内容：

{"name":"lili","url":"/iphone.html","time":"2020-09-01 11:00:00"}

注意：用户名也是：lili，但时间是 2020-09-01 11:00:00

执行结果

1> UserLog{name='lili', url='/dior.html', time=Tue Sep 01 13:00:00 CST 2020}

解释：由于第三次输入数据的 time 较早，所以输出结果仍然是上一次的结果

原文出处：/show_1IX43Red6x39.html