Flink DataStream Java Lambda 表达式的限制

Posted SunnyYoona

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Flink DataStream Java Lambda 表达式的限制相关的知识,希望对你有一定的参考价值。


原文:Flink DataStream Java Lambda 表达式的限制

Java 8 引入了一些新的语言特性,旨在实现更快、更清晰的编码。凭借最重要的特性 ‘Lambda 表达式’,打开了函数式编程的大门。Lambda 表达式允许以直接的方式实现和传递函数,而无需声明额外的(匿名)类。Flink 支持对 Java API 的所有算子使用 Lambda 表达式,但是当 Lambda 表达式使用 Java 泛型时,我们都必须显式声明类型信息。

下面通过两个例子来展示如何使用 Lambda 表达式并描述使用上的限制。

1. MapFunction与泛型

以下示例说明了如何实现一个简单的 map() 函数,该函数使用 Lambda 表达式对其输入进行平方,输出值为本身和平方的元组:

env.fromElements(1, 2, 3)
  .map(i -> Tuple2.of(i, i*i))
  .print();

由于在 map 函数中使用了 Lambda 表达式,结果抛出如下异常信息:

从上面报错信息可以知道 Lambda 中由于使用了泛型导致类型擦除而抛异常。map() 函数返回值 Tuple2<Integer, Integer> 被擦除为 Tuple2 map(Integer value)。在许多情况下,当涉及 Java 泛型时,Lambda 方法无法为自动类型提取提供足够的信息。报错信息中也给出了解决方案:

  • 使用一个(匿名)类来代替实现 ‘org.apache.flink.api.common.functions.MapFunction’ 接口。
  • 必须使用类型信息显式指定类型。

下面我们详细看一下具体的对应方案。

1.1 匿名内部类方式

使用匿名内部类来代替实现 ‘org.apache.flink.api.common.functions.MapFunction’ 接口:

env.fromElements(1, 2, 3)
    .map(new MapFunction<Integer, Tuple2<Integer, Integer>>() {
        @Override
        public Tuple2<Integer, Integer> map(Integer i) throws Exception {
            return new Tuple2<Integer, Integer>(i, i * i);
        }
    })
    .print();

为什么采用匿名内部类就没有问题?因为匿名内部类会编译成相关的类字节码存储在 class 文件中,而 Lambda 表达式只是 Java 的语法糖并不会存在相关的类字节码,Lambda 表达式是在运行时调用 invokedynamic 指令,亦即在第一次执行其逻辑时才会确定。因此 Lambda 表达式比起匿名内部类,会丢失更多的类型信息。

1.2 自定义类方式

使用自定义类来代替实现 ‘org.apache.flink.api.common.functions.MapFunction’ 接口:

public static class MyMapFunction implements MapFunction<Integer, Tuple2<Integer, Integer>> {
    @Override
    public Tuple2<Integer, Integer> map(Integer i) {
        return Tuple2.of(i, i*i);
    }
}

env.fromElements(1, 2, 3)
  .map(new MyMapFunction())
  .print();

1.3 returns方式

使用 returns 语句显示的指明类型信息:

env.fromElements(1, 2, 3)
    .map(i -> Tuple2.of(i, i*i))
    .returns(Types.TUPLE(Types.INT, Types.INT))
    .print();

2. FlatMap与泛型

在下面示例中我们在 flatMap() 函数中使用 Lambda 表达式将字符串拆成多行:

env.fromElements("1,2,3", "4,5")
    .flatMap((String input, Collector<String> out) -> {
        String[] params = input.split(",");
        for(String value : params) {
            out.collect(value);
        }
    })
    .print();

由于 flatMap() 函数签名 void flatMap(IN value, Collector out) 中有泛型 Collector,所以在编译是进行泛型类型擦除,最终编译为 flatMap(IN value, Collector out)。这使得 Flink 无法自动推断输出类型的类型信息。所以在 flatMap() 函数中使用 Lambda 表达式会抛出类似如下的异常:

跟 MapFunction 的报错信息基本一致,都是由于 Lambda 中使用泛型导致类型擦除。解决方案也类似:

  • 使用一个(匿名)类来代替实现 ‘org.apache.flink.api.common.functions.FlatMapFunction’ 接口。
  • 必须使用类型信息显式指定类型。

下面我们详细看一下具体的对应方案。

2.1 匿名内部类方式

使用匿名内部类来代替实现 ‘org.apache.flink.api.common.functions.FlatMapFunction’ 接口:

env.fromElements("1,2,3", "4,5")
    .flatMap(new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String input, Collector<String> out) throws Exception {
            String[] params = input.split(",");
            for(String value : params) {
                out.collect(value);
            }
        }
    })
    .print();

2.2 自定义类方式

使用自定义类来代替实现 ‘org.apache.flink.api.common.functions.FlatMapFunction’ 接口:

public static class MyFlatMapFunction implements FlatMapFunction<String, String> {
    @Override
    public void flatMap(String input, Collector<String> out) throws Exception {
        String[] params = input.split(",");
        for(String value : params) {
            out.collect(value);
        }
    }
}

env.fromElements("1,2,3", "4,5")
  .flatMap(new MyFlatMapFunction())
  .print();

2.3 returns方式

使用 returns 语句显示的指明类型信息:

env.fromElements("1,2,3", "4,5")
  .flatMap((String input, Collector<String> out) -> {
      String[] params = input.split(",");
      for(String value : params) {
          out.collect(value);
      }
  })
  .returns(String.class)
  .print();

欢迎关注我的公众号和博客:

参考:

以上是关于Flink DataStream Java Lambda 表达式的限制的主要内容,如果未能解决你的问题,请参考以下文章

Flink DataStream Java Lambda 表达式的限制

第三章 flink流处理API - map和flatmap

Flink DataStream API

Flink1.14实战Flink DataStream 算子 FlatMap

Flink 源码解读

Flink 源码解读