Flink

Java Lambda 表达式

Sat Aug 22, 2020

Java Lambda Expressions

Java Lambda 表达式 #

Java 8 引入了一些新的语言功能，旨在实现更快、更清晰的编码。其中最重要的功能是所谓的"Lambda 表达式"，它打开了函数式编程的大门。Lambda 表达式允许以一种直接的方式实现和传递函数，而无需声明额外的（匿名）类。

注意 Flink 支持对 Java API 的所有操作符使用 lambda 表达式，但是，每当 lambda 表达式使用 Java 属的时候，你需要明确地声明类型信息。

本文档展示了如何使用 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 map(IN value) 的实现中自动提取结果类型信息，因为 OUT 不是通用的，而是 Integer。

遗憾的是，像 flatMap() 这样签名为 void flatMap(IN value, Collector<OUT> out) 的函数被 Java 编译器编译成 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;
    }
}

原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/java_lambdas.html