算子学习KeyBy Reduce Process

接着上文，这个做简单的算子学习

自定义source

先上实体类

@Getter
@Setter
@ToString
@AllArgsConstructor
@NoArgsConstructor
public class WaterMarkData {
    private int id;  // id
    private Long ts; // 时间戳
    private int sex; // 0 男 1 女
    private int age; // 年龄
}

自定义source数据产生

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.flink.demo.module.WaterMarkData;  //注意实体包路径

import java.util.Random;

public class DataSource implements SourceFunction<WaterMarkData> {

    private volatile boolean flag = true;

    @Override
    public void run(SourceContext<WaterMarkData> ctx) throws Exception {

        while (flag) {
            WaterMarkData ws = new WaterMarkData(
                    new Random().nextInt(100000),
                    System.currentTimeMillis(),
                    new Random().nextInt(1),
                    new Random().nextInt(100)

            );
            ctx.collect(ws);
            Thread.sleep(500);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}

flink代码数据处理

代码可以直接运行

import org.apache.flink.api.common.functions.ReduceFunction;
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.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.flink.demo.common.source.DataSource;
import org.flink.demo.module.WaterMarkData;

public class Flink03_Transform_Key {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSource dataSource = new DataSource();

        DataStreamSource<WaterMarkData> markDataDataStreamSource = env.addSource(dataSource);

        // keyBy函数  reduce返回其中年龄最小的
        KeyedStream<WaterMarkData, Integer> markDataIntegerKeyedStream = markDataDataStreamSource.keyBy(WaterMarkData::getSex);

        SingleOutputStreamOperator<WaterMarkData> markDataSingleOutputStreamOperator = markDataIntegerKeyedStream.reduce(new ReduceFunction<WaterMarkData>() {
            //value1 value2代表keyby后分区内的数据信息
            @Override
            public WaterMarkData reduce(WaterMarkData value1, WaterMarkData value2) throws Exception {
                if (value1.getAge() < value2.getAge()) {
                    return value1;
                } else {
                    return value2;
                }
            }
        });

        markDataSingleOutputStreamOperator.print();

        /**
         * reduce 是分区内数据的聚合 process是每个数据的处理 这里直接process输出
         */
        SingleOutputStreamOperator<WaterMarkData> waterMarkDataSingleOutputStreamOperator = markDataIntegerKeyedStream.process(new KeyedProcessFunction<Integer, WaterMarkData, WaterMarkData>() {
            /**
             * @param value 流到的数据信息
             * @param ctx 当前流环境 （时间，测流，定时器，都在这里）
             * @param out 数据采集器对外输出数据
             * @throws Exception
             */
            @Override
            public void processElement(WaterMarkData value, KeyedProcessFunction<Integer, WaterMarkData, WaterMarkData>.Context ctx, Collector<WaterMarkData> out) throws Exception {

                out.collect(value);
            }
        });
        waterMarkDataSingleOutputStreamOperator.print();
        env.execute();
    }
}