算子学习Map FlatMap

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

map算子学习

map是对每个元素处理，一进一出

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink02_Transform {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //直接从数据中读取数据
        DataStreamSource<Integer> envStream = env.fromElements(1, 2, 3, 4, 5, 6);

        //算子学习 map
        //如果逻辑简单可以直接按照lanbada写
        //1
        SingleOutputStreamOperator<Integer> map = envStream.map(data -> {
            return data * 2;
        });
        //2
        SingleOutputStreamOperator<Integer> map2 = envStream.map(data -> data * 2);
        // 两种方案都是可行
        //map函数内部是要求实现MapFunction<IN, OUT>   in输入类型 out输出类型
        SingleOutputStreamOperator<String> map3 = envStream.map(
                new MapFunction<Integer, String>() {
                    @Override
                    public String map(Integer value) throws Exception {
                        //这里可以做更加复杂的处理
                        return value.toString();
                    }
                });

        //更加复杂的情况是需要map对开不数据处理这时候可以使用RichMapFunction
        SingleOutputStreamOperator<String> map4 = envStream.map(
                new RichMapFunction<Integer, String>() {
                    //open方法做数据初始化

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //在这里可以做外部信息获取比如链接
                        super.open(parameters);
                        System.out.println("这里的代码会在map之前处理");
                    }

                    @Override
                    public String map(Integer value) throws Exception {
                        //这里可以做更加复杂的处理
                        System.out.println("正常处理逻辑代码");
                        return value.toString();
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        System.out.println("程序关闭时候");
                    }
                });

        envStream.print();
        env.execute();
    }
}

FlatMap使用

flatmap 扁平化处理，一进多出，例如list到一个个元素

import org.apache.flink.api.common.functions.FlatMapFunction;
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.util.Collector;

import java.util.Arrays;
import java.util.List;

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


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

        final DataStreamSource<List<Integer>> numberDS =
                env.fromElements(
                        Arrays.asList(1,2,3,4,5,6)
                );

        // 将一个整体拆分成个体的操作，称之位扁平化
        SingleOutputStreamOperator<Integer> flatMapDS = numberDS.flatMap(
            new FlatMapFunction<List<Integer>, Integer>() {
                @Override
                public void flatMap(List<Integer> numbers, Collector<Integer> out) throws Exception {
                    // 将每一条数据取出来放置到采集器中
                    for (Integer num : numbers) {
                        out.collect(num * 2);
                    }
                }
            }
        );

        SingleOutputStreamOperator<Integer> flatMapDS1 = numberDS.flatMap(
            (nums, out) -> {
                // 将每一条数据取出来放置到采集器中
                for (Integer num : nums) {
                    out.collect(num * 2);
                }
            }
        );
        //flatmap 相同的也有RichFlatMapFunction
        flatMapDS.print();

        env.execute();
    }
}