（十五）Flink Datastream API 编程指南 Java Lambda表达式

Java 8引入了一些新的语言特性，旨在更快、更清晰地编码。通过最重要的特性，所谓的“Lambda表达式”，它打开了函数式编程的大门。Lambda表达式允许以一种直接的方式实现和传递函数，而无需声明其他(匿名)类。

Flink支持对Java API的所有操作符使用lambda表达式，但是，每当lambda表达式使用Java泛型时，您需要显式声明类型信息。
举个栗子,看下returns如何使用：

package com.stream.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class WCDemo {
    public static void main(String[] args) throws Exception {
        //创建流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(ParameterTool.fromArgs(args));
        env.fromElements(WORDS)
                .flatMap((FlatMapFunction>) (s, collector) -> {
                    String[] s1 = s.split("\W+");

                    for (String word : s1) {
                        if (word.length() > 0) {
                            collector.collect(new Tuple2<>(word.trim(), 1));
                        }
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(one -> one.f0)
                .reduce((ReduceFunction>) (stringIntegerTuple2, t1) -> new Tuple2<>(stringIntegerTuple2.f0, stringIntegerTuple2.f1 + stringIntegerTuple2.f1))
                .print();

        env.execute("start ...");
    }

    private static final String[] WORDS = new String[]{
            "To be , or not to be , -- that is the question : --",
            "whether 'tis nobler in the mind to suffer'"
    };

}

本文档展示了如何使用lambda表达式，并描述了当前的限制。关于Flink API的一般介绍，请参阅DataSteam API概述。

下面的示例说明了如何实现一个简单的内联map()函数，该函数使用lambda表达式对其输入进行平方。map()函数的输入i和输出参数类型不需要声明，因为Java编译器可以推断它们。

env.fromElements(1, 2, 3)
// returns the squared i
.map(i -> i*i)
.print();

Flink可以自动从方法签名OUT映射(IN值)的实现中提取结果类型信息，因为OUT不是泛型，而是Integer。

不幸的是，像flatMap()这样签名为void flatMap(IN值，Collector OUT)的函数会被Java编译器编译成void flatMap(IN值，Collector OUT)。这使得Flink无法自动推断输出类型的类型信息。

Flink很可能会抛出类似如下的异常:

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.

在这种情况下，需要显式指定类型信息，否则输出将被视为Object类型，这将导致低效的序列化。

DataStream input = env.fromElements(1, 2, 3);

// collector type must be declared
input.flatMap((Integer number, Collector out) -> {
    StringBuilder builder = new StringBuilder();
    for(int i = 0; i < number; i++) {
        builder.append("a");
        out.collect(builder.toString());
    }
})
// provide type information explicitly
.returns(Types.STRING)
// prints "a", "a", "aa", "a", "aa", "aaa"
.print();

使用泛型返回类型的map()函数时也会出现类似的问题。在下面的例子中，方法签名Tuple2 map(Integer值)被擦除为Tuple2 map(Integer值)。

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

env.fromElements(1, 2, 3)
    .map(i -> Tuple2.of(i, i))    // no information about fields of Tuple2
    .print();

一般来说，这些问题可以通过多种方式解决:

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;

// use the explicit ".returns(...)"
env.fromElements(1, 2, 3)
    .map(i -> Tuple2.of(i, i))
    .returns(Types.TUPLE(Types.INT, Types.INT))
    .print();

// use a class instead
env.fromElements(1, 2, 3)
    .map(new MyTuple2Mapper())
    .print();

public static class MyTuple2Mapper extends MapFunction> {
    @Override
    public Tuple2 map(Integer i) {
        return Tuple2.of(i, i);
    }
}

// use an anonymous class instead
env.fromElements(1, 2, 3)
    .map(new MapFunction> {
        @Override
        public Tuple2 map(Integer i) {
            return Tuple2.of(i, i);
        }
    })
    .print();

// or in this example use a tuple subclass instead
env.fromElements(1, 2, 3)
    .map(i -> new DoubleTuple(i, i))
    .print();

public static class DoubleTuple extends Tuple2 {
    public DoubleTuple(int f0, int f1) {
        this.f0 = f0;
        this.f1 = f1;
    }
}