首页 > 编程语言 >55、Flink 中使用 Java Lambda 表达式详解

55、Flink 中使用 Java Lambda 表达式详解

时间:2024-07-03 10:00:47浏览次数:23  
标签:map flink Java fromElements 55 Flink Tuple2 env print

1)概述
1.注意

Flink 支持对 Java API 的所有算子使用 Lambda 表达式,但是,当 Lambda 表达式使用 Java 泛型时,需要 显式 地声明类型信息。

2.示例和限制

示例: map() 函数使用 Lambda 表达式计算输入值的平方。

不需要声明 map() 函数的输入 i 和输出参数的数据类型,因为 Java 编译器会对它们做出推断。

env.fromElements(1, 2, 3)
// 返回 i 的平方
.map(i -> i*i)
.print();

由于 OUTInteger 而不是泛型,所以 Flink 可以从方法签名 OUT map(IN value) 的实现中自动提取出结果的类型信息。

但像 flatMap() 这样的函数,它的签名 void flatMap(IN value, Collector 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 类型,这会导致低效的序列化。

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

// 必须声明 collector 类型
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());
    }
})
// 显式提供类型信息
.returns(Types.STRING)
// 打印 "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))    // 没有关于 Tuple2 字段的信息
    .print();

解决方式如下

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

// 使用显式的 ".returns(...)"
env.fromElements(1, 2, 3)
    .map(i -> Tuple2.of(i, i))
    .returns(Types.TUPLE(Types.INT, Types.INT))
    .print();

// 使用类来替代
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);
    }
}

// 使用匿名类来替代
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();

// 也可以像这个示例中使用 Tuple 的子类来替代
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;
    }
}

标签:map,flink,Java,fromElements,55,Flink,Tuple2,env,print
From: https://blog.csdn.net/m0_50186249/article/details/140144288

相关文章

  • 56、Flink DataStream 的管理执行配置详解
    1)概述1.执行配置StreamExecutionEnvironment包含了ExecutionConfig,它允许在运行时设置作业特定的配置值。StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();ExecutionConfigexecutionConfig=env.getConfig();以下是可用......
  • 从0到1Flink的成长之路(二十)-Flink 高级特性(二)之自动重启策略和恢复 ,固定延迟重启策
    从0到1Flink的成长之路(二十)-Flink高级特性(二)之自动重启策略和恢复,,固定延迟重启策略(开发中使用)自动重启策略和恢复1)、重启策略配置方式配置文件在flink-conf.yml中可以进行配置,示例如下:restart-strategy:fixed-delayrestart-strategy.fixed-delay.attempts:3restart-strat......
  • 用JavaScript来优化数独验证的过程
    问题陈述给定一个9x9数独棋盘,确定它是否有效。棋盘由一个二维数组表示,其中空单元格用表示'.'。有效的数独棋盘满足以下规则:每行必须包含数字1–9,且不能重复。每列必须包含数字1–9,且不能重复。九个3x3子网格中的每一个都必须包含数字1–9,且不能重复。初步方法一种简......
  • 有手就会的 Java 处理压缩文件
    @目录前言背景第一步:编写代码1.1请求层1.2业务处理层1.3新增配置第二步:解压缩处理2.1引入依赖2.2解压缩工具类总结前言请各大网友尊重本人原创知识分享,谨记本人博客:南国以南i、提示:以下是本篇文章正文内容,下面案例可供参考背景在项目出现上传文件,其中文件包含压缩包,......
  • WebStorm 2024 for Mac JavaScript前端开发工具
    Mac分享吧文章目录效果一、下载软件二、开始安装1、双击运行软件(适合自己的M芯片版或Intel芯片版),将其从左侧拖入右侧文件夹中,等待安装完毕2、应用程序显示软件图标,表示安装成功3、打开访达,点击【文稿】。将安装包内的【ja-netfilter】文件夹拖到文稿中4、填写内容,修改用......
  • 1367java jsp SSM留学生交流互动论坛网站系统经验分享计划分享软件推荐网址推荐标签分
     项目技术:SSM+Maven+Vue等等组成,B/S模式+Maven管理等等。环境需要1.运行环境:最好是javajdk1.8,我们在这个平台上运行的。其他版本理论上也可以。2.IDE环境:IDEA,Eclipse,Myeclipse都可以。推荐IDEA;3.tomcat环境:Tomcat7.x,8.x,9.x版本均可4.硬件环境:windows7/8/1......
  • 555、基于51单片机的汽车灯控制器设计(刹车、倒车、雾霾)
    完整资料或定制滴滴我(有偿)见文末。目录一、设计功能二、Proteus仿真三、原理图四、程序源码五、资料包括一、设计功能汽车灯控制器设计要求:1、汽车车尾左右两侧各有四盏灯:黄灯、红灯、雾灯、倒车照明灯,前面有照明灯(远光、近光)、黄灯、雾灯2、白天正常行驶时照......
  • 287java jsp SSM旅游信息网站系统旅游景点管理(源码+文档+PPT+运行视频+讲解视频)
     项目技术:SSM+Maven+Vue等等组成,B/S模式+Maven管理等等。环境需要1.运行环境:最好是javajdk1.8,我们在这个平台上运行的。其他版本理论上也可以。2.IDE环境:IDEA,Eclipse,Myeclipse都可以。推荐IDEA;3.tomcat环境:Tomcat7.x,8.x,9.x版本均可4.硬件环境:windows7/8/1......
  • 554、基于51单片机的跑步机(计价,4挡)
    完整资料或定制滴滴我(有偿)见文末。目录一、设计功能二、Proteus仿真三、原理图四、程序源码五、资料包括一、设计功能跑步机计价器1、使用直流电机模拟跑步机运行2、设置4个速度档位,用户可以选择不同速度体验3、具有计费功能,单价可调二、Proteus仿真......
  • 1187java jsp SSM列车火车高铁票务信息管理系统(源码+文档+PPT+运行视频+讲解视频)
     项目技术:SSM+Maven+Vue等等组成,B/S模式+Maven管理等等。环境需要1.运行环境:最好是javajdk1.8,我们在这个平台上运行的。其他版本理论上也可以。2.IDE环境:IDEA,Eclipse,Myeclipse都可以。推荐IDEA;3.tomcat环境:Tomcat7.x,8.x,9.x版本均可4.硬件环境:windows7/8/1......