看完了Flink的datasource、sink，也就把一头一尾给看完了，从数据流入到数据流出，缺少了中间的处理环节。

而flink的大头恰恰是只在这个中间环节，如下图：

source-transform-sink-update.png

中间的处理环节比较复杂，现在也就看了其中一部分，这里先开始讲其中最简单 也最常用的map、flatmap及filter。

map

flink中dataSourceStream和java8中的map很类似，都是用来做转换处理的，看下map的实现：

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

可以看到：

1、返回的是SingleOutputStreamOperator泛型，这是个基础的类型，好多DataStream的方法都返回它，比如map、flapmap、filter、process等

2、最终是调用transform方法来实现的，看下transfrom的实现：

    @PublicEvolving
    public <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
        
        this.transformation.getOutputType();
        
        OneInputTransformation<T, R> resultTransform = new OneInputTransformation(this.transformation, operatorName, operator, outTypeInfo, this.environment.getParallelism());
        
        SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(this.environment, resultTransform);
        
        this.getExecutionEnvironment().addOperator(resultTransform);
        
        return returnStream;
    }

额，好像还不如不看，直接看怎么用吧！

@Slf4j
public class KafkaUrlSinkJob {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("zookeeper.connect", "localhost:2181");
        properties.put("group.id", "metric-group");
        properties.put("auto.offset.reset", "latest");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        SingleOutputStreamOperator<UrlInfo> dataStreamSource = env.addSource(
                new FlinkKafkaConsumer010<String>(
                        "testjin",// topic
                        new SimpleStringSchema(),
                        properties
                )
        ).setParallelism(1)
                // map操作，转换，从一个数据流转换成另一个数据流，这里是从string-->UrlInfo
                .map(string -> JSON.parseObject(string, UrlInfo.class))
           
       
    }

可以看到，kafka中传递的是String类型，在这里通过map转换后，变SingleOutputStreamOperator<UrlInfo> 类型，否则就是SingleOutputStreamOperator<String> 。

map方法不允许缺少数据，也就是原来多少条数据，处理后依然是多少条数据，只是用来做转换。

flatmap

flatmap，也就是将嵌套集合转换并平铺成非嵌套集合。看个例子，还是用上面的kafka datasource：

        // 构造一个嵌套的数据
        SingleOutputStreamOperator<List<UrlInfo>> listDataStreaamSource = dataStreamSource
                .map(urlInfo -> {
                    List<UrlInfo> list = Lists.newArrayList();

                    list.add(urlInfo);

                    UrlInfo urlInfo1 = new UrlInfo();
                    urlInfo1.setUrl(urlInfo.getUrl() + "-copy");
                    urlInfo1.setHash(DigestUtils.md5Hex(urlInfo1.getUrl()));

                    list.add(urlInfo1);

                    return list;
                }).returns(new ListTypeInfo(UrlInfo.class));

        listDataStreaamSource.addSink(new PrintSinkFunction<>());

说明：

1、注意这里的returns方法，如果不指定，会在运行时报错

/*I think the short description of the error message is quite good, but let me expand it a bit.
In order to execute a program, Flink needs to know the type of the values that are processed because it needs to serialize and deserialize them. Flink's type system is based on TypeInformation which describes a data type. When you specify a function, Flink tries to infer the return type of that function. In case of the FlatMapFunction of your example the type of the objects that are passed to the Collector.
Unfortunately, some Lambda functions lose this information due to type erasure such that Flink cannot automatically infer the type. Therefore, you have to explicitly declare the return type.

如果直接上面这样转换，因为lambda表达式会丢失部分信息，会报如下异常：
org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing. In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved. An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface. Otherwise the type has to be specified explicitly using type information.
*/

不过由于返回的是一个List<Ojbect>,不可能直接用 List<Object>.class，没这种写法。而flink则

提供了更多选项，这里使用的是

public SingleOutputStreamOperator<T> returns(TypeInformation<T> typeInfo){}

这个构造函数，而ListTypeInfo则是继承TypeInfomation抽象类的一个List实现。

和上文的KafkaSender一起运行，会有如下结果：

kafkaSender:

2019-01-15 20:21:46.650 [main] INFO  org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : e89bffd6b2eff799
2019-01-15 20:21:46.653 [main] INFO  myflink.KafkaSender - send msg:{"domain":"so.com","id":0,"url":"http://so.com/1547554906650"}

KafkaUrlSinkJob

[UrlInfo(id=0, url=http://so.com/1547554906650, hash=null), UrlInfo(id=0, url=http://so.com/1547554906650-copy, hash=efb0862d481297743b08126b2cda602e)]

也就是一个UrlInfo 扩展成了 一个List<UrlInfo>

下面看看怎么使用flatmap

...
    
SingleOutputStreamOperator<UrlInfo> flatSource = listDataStreaamSource.flatMap(new FlatMapFunction<List<UrlInfo>, UrlInfo>() {
            @Override
            public void flatMap(List<UrlInfo> urlInfos, Collector<UrlInfo> collector) throws Exception {
                urlInfos.parallelStream().forEach(urlInfo -> collector.collect(urlInfo));
            }
        });

flatSource.addSink(new PrintSinkFunction<>());

...

当然可以写成lambda表达式：（注意lambda表达式需要显式指定return type）

SingleOutputStreamOperator<UrlInfo> flatSource = listDataStreaamSource.flatMap(
                (FlatMapFunction<List<UrlInfo>, UrlInfo>) (urlInfos, collector) ->
                        urlInfos.parallelStream().forEach(urlInfo -> collector.collect(urlInfo))).returns(UrlInfo.class);

看看打印出来的结果：

2> [UrlInfo(id=0, url=http://so.com/1547554906650, hash=null), UrlInfo(id=0, url=http://so.com/1547554906650-copy, hash=efb0862d481297743b08126b2cda602e)]
1> [UrlInfo(id=0, url=http://so.com/1547554903640, hash=null), UrlInfo(id=0, url=http://so.com/1547554903640-copy, hash=138f79ecc92744a65b03132959da2f73)]

1> UrlInfo(id=0, url=http://so.com/1547554903640-copy, hash=138f79ecc92744a65b03132959da2f73)
1> UrlInfo(id=0, url=http://so.com/1547554903640, hash=null)
2> UrlInfo(id=0, url=http://so.com/1547554906650, hash=null)
2> UrlInfo(id=0, url=http://so.com/1547554906650-copy, hash=efb0862d481297743b08126b2cda602e)

也就是说，flatmap方法最终返回的是一个collector，而这个collector只有一层，当输入数据有嵌套的情况下，可以将数据平铺处理。

当然，不只是针对嵌套集合，由于flatmap返回的数据条数并不会做限制，也就可以做一些扩展数据处理的情况，如下：

dataStream.flatMap((FlatMapFunction<String, String>) (value, out) -> {
            for (String word : value.split(" ")) {
                out.collect(word);
            }
        });

这里就是将string使用空格切割后，组成一个新的dataStream.

filter

顾名思义，filter用于过滤数据，继续在上面代码的基础上写测试。为了避免干扰，将上面两个dataSourceStream.addSink注释掉，添加以下代码：

// 根据domain字段，过滤数据，只保留BAIDU的domain
SingleOutputStreamOperator<UrlInfo> filterSource = flatSource.filter(urlInfo -> {
            if(StringUtils.equals(UrlInfo.BAIDU,urlInfo.getDomain())){
                return true;
            }
            return false;
        });

        filterSource.addSink(new PrintSinkFunction<>());

这里排除别的domain数据，只保留BAIDU的数据，运行结果就不贴出来了，验证了filter的效果。