Java Lambda表达式
- 示例和限制

Java Lambda表达式

译者：flink.sojb.cn

Java 8引入了几种新的语言函数，旨在实现更快，更清晰的编码。凭借最重要的函数，即所谓的“Lambda表达式”，它打开了函数式编程的大门。Lambda表达式允许以直接的方式实现和传递函数，而无需声明其他（匿名）类。

注意 Flink支持对Java API的所有算子使用lambda表达式，但是，每当lambda表达式使用Java泛型时，您需要显式声明类型信息。

本文档介绍如何使用lambda表达式并描述当前的限制。有关Flink API的一般介绍，请参阅编程指南

示例和限制

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

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

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

遗憾的是，Java编译器会编译flatMap()带有签名的函数。这使得Flink无法自动推断输出类型的类型信息。void flatMap(IN value, Collector<OUT> out)``void flatMap(IN value, Collector out)

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.

在这种情况下，Required显式指定类型信息，否则输出将被视为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, 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;
    }
}