Java Lambda Expressions

Java 8引入了几个新的语言特性，旨在实现更快、更清晰的编码。最重要的特性，即所谓的“Lambda表达式”，为函数式编程打开了大门。Lambda表达式允许以简单的方式实现和传递函数，而不必声明其他(匿名)类。

注意Flink支持JavaAPI的所有操作符使用lambda表达式，但是，每当lambda表达式使用Java泛型时，都需要声明信息类型 explicitly。

本文档演示如何使用lambda表达式，并描述当前的限制。有关FLinkAPI的一般介绍，请参阅编程Guide]

Examples and Limitations （实施例和限制）

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

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

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

不幸的是，由Java编译器将诸如flatMap()和签名void flatMap(IN value, Collector<OUT> out) 之类的函数编译为void flatMap(IN value, 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类型，这将导致无效的序列化。

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.util.Collector;
DataSet<Integer> input = env.fromElements(1, 2, 3);
// collector type must be declared
input.flatMap((Integer number, Collector<String> 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<Integer, Integer> map(Integer value) 删除为 Tuple2 map(Integer value) 。

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<Integer, Tuple2<Integer, Integer>> {
    @Override
    public Tuple2<Integer, Integer> map(Integer i) {
        return Tuple2.of(i, i);
    }
}
// use an anonymous class instead
env.fromElements(1, 2, 3)
    .map(new MapFunction<Integer, Tuple2<Integer, Integer>> {
        @Override
        public Tuple2<Integer, Integer> 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<Integer, Integer> {
    public DoubleTuple(int f0, int f1) {
        this.f0 = f0;
        this.f1 = f1;
    }
}