flink1.12.x Transformation-reduce自定义聚合

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

说明

有些时候，flink提供 max()、min()、sum() 等聚合函数，无法满足需求，我们需要自定义聚合，这时，就需要 reduce()

两种方式

  • 正常的reduce
  • window 窗口的 reduce

应用场景

利用reduce函数来实时统计每种商品的商品数量

api

keyBy()

调用 keyBy() 后，返回 KeyedStream 类型

KeyedStream

该类有聚合函数，如：max()、min()、sum()、reduce() 等

reduce() 方法

源码：

SingleOutputStreamOperator<T> reduce(ReduceFunction<T> reducer)

泛型：传入数据的类型

形参：需要传入 ReduceFunction 类型

ReduceFunction

源码如下：

// 根据下面注解可知，支持lambda表达式
@FunctionalInterface
@Public
// 只有一个泛型
public interface ReduceFunction<T> extends Function, Serializable {
    /*
    * 传入的2个元素，返回类型与传入类型相同
    * 第一次执行：var1 是第一个元素，var2 是第二个元素，计算后得出结果（该结果也称之为状态）
    * 之后执行：var1 是上一次的计算结果，var2 是第三个元素，计算后得出结果（该结果也称之为状态）
    */
    T reduce(T var1, T var2) throws Exception;
}

状态：聚合之后的 结果，称之为 状态

案例

根据用户名分组，统计各个用户访问次数，用 reduce() 方法实现 flink1.12.x Transformation-sum()求和 中的功能

主类

package reduce;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
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.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 scala.Tuple3;

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);

        // 将json转换成，
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.map(s -> {
            ObjectMapper mapper = new ObjectMapper();
            UserLog result = mapper.readValue(s, UserLog.class);
            return Tuple2.of(result.getName(),1);
        }).returns(Types.TUPLE(Types.STRING,Types.INT));

        KeyedStream<Tuple2<String,Integer>, String> ks = mapSOSO.keyBy(item -> item.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> reduceSOSO = ks.reduce(
                (item1, item2) -> Tuple2.of(item1.f0, item1.f1 + item2.f1)
        );

        reduceSOSO.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)

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

原文出处：/show_1IX43doNvXFT.html

吉ICP备15003043号