flink1.12.x Transformation-sum()求和

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

说明

在 flink1.12.x Transformation-keyBy算子 中，执行 keyBy 分组后，可以执行 sum() 求和

案例

根据用户名分组，统计各个用户访问次数

主类

package keyBy;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
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 求和sum {
    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);


        // 由于要统计次数，所以将数据转成 Tuple2<String,Long> ，第一个元素是用户名，第二个元素是1
        SingleOutputStreamOperator<Tuple2<String,Long>> mapSOSO = dss.map(s -> {
            ObjectMapper mapper = new ObjectMapper();
            UserLog result = mapper.readValue(s, UserLog.class);
            return Tuple2.of(result.getName(),1L);
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));// 泛型擦除，所以指定返回类型
        // 根据 Tuple2<String,Long> 第一个元素（用户名）进行分组（keyBy），然后对第二个元素（即：1）执行 sum()求和
        SingleOutputStreamOperator<Tuple2<String,Long>> sumSOSO = mapSOSO
                .keyBy(item -> item.f0)
                .sum(1);

        sumSOSO.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":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
{"name":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
{"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":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00"}

执行结果

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

4> (lucy,1)
4> (lucy,2)
1> (lili,1)
4> (lucy,3)

解释：根据姓名分组，然后求和

执行过程：

  • 不论哪个节点 收到第一个 lucy，打印 (lucy,1)；
  • 不论哪个节点 收到第二个lucy，打印 (lucy,2)；
  • 不论哪个节点 收到第三个 lucy，都会打印 (lucy,3)

保证了当前统计值是实时 全局增加的

原文出处：/show_1IX43odGLjkJ.html

吉ICP备15003043号