FLink12--KeyByReduceApp

一、依赖

参考博文：https://www.cnblogs.com/robots2/p/16048648.html

二、代码

package net.xdclass.class9;

import java.util.Date;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
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 net.xdclass.model.VideoOrder;

/**
 * @desc reduce算子,和sum类似，sum做简单聚合，reduce做复杂聚合
 * aggregate支持更复杂聚合
 * @menu
 */
public class FLink12KeyByReduceApp {

    public static void main(String[] args) throws Exception{
        //WebUi方式运行
//        final StreamExecutionEnvironment env =
//                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置运行模式为流批一体
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //并行度
        env.setParallelism(1);
        //设置为自定义source
//        DataStream ds = env.addSource(new VideoOrderSourceV2());
        DataStream ds = env.fromElements(
                new VideoOrder("20190242812", "springboot教程", 10,1001, new Date()),
                new VideoOrder("20194350812", "微服务SpringCloud", 20,1001, new Date()),
                new VideoOrder("20190814232", "Redis教程", 30,1001, new Date()),
                new VideoOrder("20190523812", "??开发教程", 40,1001, new Date()),
                new VideoOrder("201932324", "百万并发实战Netty", 50,1001, new Date()),
                new VideoOrder("20190242812", "springboot教程", 10,1001, new Date()),
                new VideoOrder("20190814232", "Redis教程", 30,1001, new Date()));


        KeyedStream videoOrderObjectKeyedStream = ds.keyBy(new KeySelector() {
            @Override
            public Object getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });

        //reduce,做聚合。合并数据，返回新的聚合对象。大于两条才会触发
        SingleOutputStreamOperator reduceResult = videoOrderObjectKeyedStream.reduce(
                new ReduceFunction() {
                    @Override
                    public VideoOrder reduce(VideoOrder value1, VideoOrder value2) throws Exception {
                        VideoOrder reduceOrder = new VideoOrder();
                        reduceOrder.setTitle(value1.getTitle());
                        reduceOrder.setMoney(value1.getMoney() + value2.getMoney());
                        return reduceOrder;
                    }
                });

        reduceResult.print();

        //DataStream需要调用execute,可以取个名称
        env.execute("reduce map job");
    }
}