第三章 flink流处理API - map和flatmap

1. Map算子

功能

map有映射的意思， 作用是可以把一个输入的数据转为另外一个数据（比如把小写字母转换为大写字母， 数字转换成他的相反数等）。

数据结构

DataStream → DataStream: 输入一个参数产生一个参数。

用法

实现MapFunction接口后，实现这个接口中的map方法， 接入参数表示输入数据， return的结果表示转换后的数据。
如： 数据翻倍等操作: dataStream.map { x => x * 2 }

源码

    public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) {
        TypeInformation<R> outType = TypeExtractor.getMapReturnTypes((MapFunction)this.clean(mapper), this.getType(), Utils.getCallLocationName(), true);
        return this.transform("Map", outType, new StreamMap((MapFunction)this.clean(mapper)));
    }

通过源码可以看到他的实际返回值是SingleOutputStreamOperator， 属于DataStream的子类。

演示

1. 代码 ： 使用map算子将数据源输入（nc -l 9000）的每一行数据都转换为大写。

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

/**
 * Created By qiuzhi. Description: Date: 2020-07-31 Time: 11:42 AM
 *
 * @author zhanghaichao
 * @date 2020/07/31
 */
public class MapDemo {

    public static void main(String[] args) throws Exception {
        int port = 9000;

        // 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 数据源
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", port, "\n");

        // 转化
        SingleOutputStreamOperator<String> streamOperator = streamSource.map(new MapOperator());

        // sink
        streamOperator.print();

        // 启动
        env.execute();
    }
}

class MapOperator implements MapFunction<String, String> {
    @Override
    public String map(String s) throws Exception {
        // 转化过程，把字符串转化为大写
        return s.toUpperCase();
    }
}

2. 输入

   
   image.png

3. 输出

   
   image.png

2. FlatMap算子

功能

flat有平坦的意思，和map结合起来表示把把输入的数据打平映射。 作用是可以把一个输入的数据转为0-N条数据（比如把一个单词中所有的字母拆出来）。

数据结构

DataStream → DataStream: 输入一个参数，产生0个、1个或者多个输出.

用法

实现FlatMapFunction接口后，实现这个接口中的flatMap方法， 第一个接入参数表示输入数据 ，第二个接入参数是一个数据收集器对象：如果希望输出该数据，就调用Collector<String>的collect将数据收集输出。
如：这个 flatmap 的功能是将句子中的单词拆分出来: dataStream.flatMap { str => str.split(" ") }

源码

       public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) {
        TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes((FlatMapFunction)this.clean(flatMapper), this.getType(), Utils.getCallLocationName(), true);
        return this.transform("Flat Map", outType, new StreamFlatMap((FlatMapFunction)this.clean(flatMapper)));
    }

通过源码可以看到他的实际返回值是SingleOutputStreamOperator， 属于DataStream的子类。

演示

1. 代码 ： 使用flatmap算子将数据源输入（nc -l 9000）的每一行数据都转换为大写之后输出每一个字母。

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;

/**
 * Created By qiuzhi. Description: Date: 2020-07-31 Time: 11:42 AM
 *
 * @author zhanghaichao
 * @date 2020/07/31
 */
public class FlatMapDemo {

    public static void main(String[] args) throws Exception {
        int port = 9000;

        // 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 数据源
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", port, "\n");

        // 转化
        SingleOutputStreamOperator<Character> streamOperator = streamSource.flatMap(new FlatMapOperator());

        // sink
        streamOperator.print();

        // 启动
        env.execute();
    }
}

class FlatMapOperator implements FlatMapFunction<String, Character> {

    @Override
    public void flatMap(String s, Collector<Character> collector) throws Exception {
        // 转换为大写
        String upperStr = s.toUpperCase();

        // 拆分每一个字母并下发
        for (int i = 0; i < upperStr.length(); i++) {
            char c = upperStr.charAt(i);
            collector.collect(c);
        }
    }
}

2. 输入

   
   image.png

3. 输出

   
   image.png