【API篇】三、转换算子API（上）

文章目录

  • 0、demo数据
  • 1、基本转换算子：映射map
  • 2、基本转换算子：过滤filter
  • 3、基本转换算子：扁平映射flatMap
  • 4、聚合算子：按键分区keyBy
  • 5、聚合算子：简单聚合sum/min/max/minBy/maxBy
  • 6、聚合算子：归约聚合reduce
  • 7、用户自定义函数：函数类
  • 8、用户自定义函数：富函数类

创建完执行环境，从数据源读入数据，就该用转换算子对数据做处理了，即使用各种转换算子，将一个或多个DataStream转换为新的DataStream

在这里插入图片描述

0、demo数据

准备一个实体类WaterSensor：

@Data
@AllArgsConstructor
@NoArgsConstructor
public class WaterSensor{private String id;   //水位传感器类型private Long ts;     //传感器记录时间戳private Integer vc;  //水位记录
}
//注意所有属性的类型都是可序列化的，如果属性类型是自定义类，那要实现Serializable接口

1、基本转换算子：映射map

map即把数据流中的数据进行转换，形成新的数据流。一一映射，消费一个元素就产出一个元素。

在这里插入图片描述
DataStream对象调用map()方法进行转换处理。map方法形参是接口MapFunction的实现对象，返回值类型还是DataStream：

public class TransMap {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_2", 2, 2));// 方式一：传入匿名类，实现MapFunctionstream.map(new MapFunction<WaterSensor, String>() {@Overridepublic String map(WaterSensor e) throws Exception {return e.id;}}).print();// 方式二：传入MapFunction的实现类stream.map(new MapFunctionImpl()).print();//方式三：Lambda表达式stream.map(t -> t.getId()).print();//方式四：Lambda表达式stream.map(WaterSensor::getId).print();env.execute();}}public class MapFunctionImpl implements MapFunction<WaterSensor, String> {@Overridepublic String map(WaterSensor e) throws Exception {return e.id;}
}

在实现MapFunction接口的时候，需要指定两个泛型，分别是输入事件和输出事件的类型，还需要重写一个map()方法，定义从一个输入事件转换为另一个输出事件的具体逻辑。当好几个作业都需要这个转换逻辑时，不用匿名内部类，而是实现类好点，省的重复写转换逻辑。

2、基本转换算子：过滤filter

通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为true则元素正常输出，若为false则元素被过滤

在这里插入图片描述
filter转换需要传入的参数需要实现FilterFunction接口，而FilterFunction内要实现filter()方法，就相当于一个返回布尔类型的条件表达式。

public class TransFilter {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3));// 传入匿名类实现FilterFunctionstream.filter(new FilterFunction<WaterSensor>() {@Overridepublic boolean filter(WaterSensor e) throws Exception {return "sensor_1".equals(e.getId());}}).print();// Lambda表达式// stream.filter(t -> "sensor_1".equals(t.getId())).print();env.execute();}}

3、基本转换算子：扁平映射flatMap

flatMap主要是将数据流中的整体拆分成一个一个的个体使用，消费一个元素，可以产生0到多个元素。先扁平化，再映射。
在这里插入图片描述

//实现：如果输入的数据是sensor_1，只打印vc; 
//如果输入的数据是sensor_2，既打印ts又打印vcpublic class TransFlatmap {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3));stream.flatMap(new FlatMapFunctionImpl()).print();env.execute();}} public class FlatMapFunctionImpl implements FlatMapFunction<WaterSensor, String> {@Overridepublic void flatMap(WaterSensor value, Collector<String> out) throws Exception {if (value.id.equals("sensor_1")) {out.collect(String.valueOf(value.vc));   //一进一出} else if (value.id.equals("sensor_2")) {out.collect(String.valueOf(value.ts));   //一进多出out.collect(String.valueOf(value.vc));}//sensor_3 一进0出}
}
//value为WaterSensor类型，收集器为String类型，即WaterSensor转String

map和flatMap相比，map总是能一进一出是因为MapFunction接口的map方法是有return返回值的，一个传入，肯定对应一个返回。而flatMap下，FlatMapFunction接口的flatMap方法返回值类型为void，最终返回啥，是靠收集器往下游传，调用n次采集器的collect方法，就输出n条数据，一次也不调，那就是不处理，又是void，那就相当于被过滤了，因此有了flatMap的一进多出：

  • 一进一出
  • 一进多出
  • 一进零出

4、聚合算子：按键分区keyBy

对海量数据进行聚合计算前，分组是必要的。

在这里插入图片描述

  • 按键分区keyBy，返回的是一个KeyedStream键控流
  • keyBy不是转换算子，不能设置并行度，只是对数据做一个重分区
  • 在内部，是通过计算key的哈希值（hash code），对分区数进行取模运算来实现的。所以这里key如果是POJO的话，必须要自己重写hashCode()方法

关于keyBy分组和分区的关系：

  • keyBy是对数据分组，保证相同key的数据在同一个分区
  • 分区，一个子任务可以理解为一个分区
  • 一个分区（子任务）中可以有多个分组
//演示以demo类的id字段来分类
public class TransKeyBy {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3));// 方式一：使用Lambda表达式KeyedStream<WaterSensor, String> keyedStream = stream.keyBy(t -> t.id);// 方式二：使用匿名类实现KeySelectorKeyedStream<WaterSensor, String> keyedStream1 = stream.keyBy(new KeySelector<WaterSensor, String>() {@Overridepublic String getKey(WaterSensor e) throws Exception {return e.id;}});//分区后继续做你需要的聚合env.execute();}
}
  • keyBy得到的结果将不再是DataStream，而是会将DataStream转换为KeyedStream
  • KeyedStream泛型中第一个为流中的元素类型外，第二个是key的类型
  • KeyedStream也继承自DataStream，所以基于它的操作也都归属于DataStream API
  • 只有基于KeyedStream才可以做后续的聚合操作（比如sum，reduce）

5、聚合算子：简单聚合sum/min/max/minBy/maxBy

注意点：

  • 在完成keyBy分组后，可以进行简单聚合
  • sum/min/max/minBy/maxBy是KeyedStream类下的API，因此必须先完成分组
  • 而简单聚合算子返回的，又变回了一个SingleOutputStreamOperator，即先分区、后聚合，得到的依然是一个DataStream
  • 是分组内的聚合，即对同一个key的数据进行聚合，不会跨key聚合

关于这些API：

  • sum()：在分组内，对指定的字段做叠加求和
  • min()：在分组内，对指定的字段求最小值
  • max()：在分组内，对指定的字段求最大值
  • minBy()：与min类似，区别是，min只计算指定字段的最小值，其他字段会保留最初第一条数据的值，而minBy则是字段最小值所在的整条数据。也就是除了指定字段，其他字段以谁为准的区别。
  • maxBy()：与max类似，区别同上
public class TransAggregation {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3));stream.keyBy(t -> t.id).max("vc");    // 指定字段名称//stream.keyBy(t -> t.id).max(2);  //报错env.execute();}
}

注意，这几个聚合算子的传参有两种：指定位置，和指定名称，对于元组类型的数据，两种都行。但如果数据流中的类型不是元组，而是一个pojo类，那就只能通过字段名来指定，而不能传一个位置，否则报错Cannot reference field by position on POJO

一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个key的数据流上。

6、聚合算子：归约聚合reduce

public class TransFilter {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<WaterSensor> stream = env.fromElements(new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_1", 3, 3),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3));// 传入匿名类实现ReduceFunctionstream.reduce(new ReduceFunction<WaterSensor>() {@Overridepublic boolean reduce(WaterSensor value1, WaterSensor value2) throws Exception {System.out.println("value1===" + value1);System.out.println("value2===" + value2);return new WaterSensor(value1.getId(),value2.getTs(),value1.getVc() + value2.getVc());}}).print();env.execute();}}

运行：

在这里插入图片描述

总结：

  • reduce算子依旧是keyBy之后KeyedStream的API

  • 该算子传入一个ReduceFunction对象，要求数据的输入类型等于输出类型
    在这里插入图片描述

  • ReduceFunction接口的reduce方法，value1和value2是流中某key分组的两个数据，中途，value1是之前的计算结果（存状态，有状态计算），value2是后面新来的数据

  • 每个key的分组里第一条数据来的时候，不会执行reduce方法，只是存起来，然后就发到下游了

  • reduce算子和前面的简单算子一样，会存每一个key的状态值，且状态不会清空，因此，如果是无界流，其key值要有限个

7、用户自定义函数：函数类

自定义函数，即用户根据自己的需求，重新实现算子的逻辑。Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等。所以用户可以自定义一个函数类，实现对应的接口。

DataStreamSource<WaterSensor> stream = env.fromElements(        new WaterSensor("sensor_1", 1, 1),new WaterSensor("sensor_1", 2, 2),new WaterSensor("sensor_2", 2, 2),new WaterSensor("sensor_3", 3, 3)
);DataStream<String> stream = stream.filter(new FilterFunctionImpl("sensor_1"));  //new对象的时候传入str,通过构造方法赋值给了id属性
public  class FilterFunctionImpl implements FilterFunction<WaterSensor> {private String id;FilterFunctionImpl(String id) { this.id=id; }@Overridepublic boolean filter(WaterSensor value) throws Exception {return thid.id.equals(value.id);   //当前对象的id属性}
}

关于函数类，写实现类、Lambda表达式、匿名内部类等方式重写算子对应的接口，前面已经演示过，上面重点改良了一下代码，把过滤关键字做为类的属性，通过构造方法传了进去。

8、用户自定义函数：富函数类

富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，可以实现更复杂的功能。

public class RichFunctionExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);env.fromElements(1,2,3,4).map(new RichMapFunction<Integer, Integer>() {@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");}@Overridepublic Integer map(Integer integer) throws Exception {return integer + 1;}@Overridepublic void close() throws Exception {super.close();System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");}}).print();env.execute();}
}

在这里插入图片描述

生命周期的方法即：

  • open()：每个子任务，在启动时，调用一次
  • close()：每个子任务，在结束时，调用一次

但需要注意：

  • 当一个算子的实际工作方法例如map()或者filter()方法被调用之前，open()会首先被调用

  • 处理有界流，处理完以后程序运行结束，调用close

  • 处理无界流，程序中止时调用close

  • 如果Flink是异常中止，则不会调用close

  • 如果是正常调用cancle命令（控制台去cancle），则会正常调用close方法

关于富函数：

  • 相比普通的自定义函数类，富函数多了一个运行时上下文对象，可通过这个对象获取到运行时环境的信息，比如子任务编号、子任务名称
  • 有的Flink函数类都有其Rich版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction等
  • 处理数据需求有时机要求时，可使用富函数

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/160651.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

maven 新建模块 导入后 按Ctrl 点不进新建模块pom定义

maven 新建模块 导入后 按Ctrl 点不进新建模块pom定义

新建的ruoyi-common-mybatisplus 模块,导入一直不正常 画出的模块一直导入不进来 这是提示信息 这是正常的提示信息 加上 <version>3.6.3</version> 后,才一切正常
阅读更多...
C++入门之引用与内联函数

C++入门之引用与内联函数

一、引用 1、初步理解 引用在语法上的理解就是起别名&#xff0c;用法就是在类型后面加&&#xff0c;例子&#xff1a;int a 1; int& b a; 上例所示&#xff0c;执行后&#xff0c;b就是a的别名&#xff0c;它们代表同一块空间&#xff0c;a的改变会影响b&#xff0…
阅读更多...
进阶JAVA篇-如何理解作为参数使用的匿名内部类与 Arrays 类的常用API（九）

进阶JAVA篇-如何理解作为参数使用的匿名内部类与 Arrays 类的常用API（九）

目录 目录 API 1.0 Arrays 类的说明 1.1 Arrays 类中的 toString() 静态方法 1.2 Arrays 类中的 copyOfRange(int[] original, int from, int to) 静态方法 1.3 Arrays 类中的 copyOf(int[] original, int newLength) 静态方法 1.4 Arrays 类中的 setAll(do…
阅读更多...
深入了解桶排序：原理、性能分析与 Java 实现

深入了解桶排序：原理、性能分析与 Java 实现

桶排序&#xff08;Bucket Sort&#xff09;是一种排序算法&#xff0c;通常用于将一组数据分割成有限数量的桶&#xff08;或容器&#xff09;&#xff0c;然后对每个桶中的数据进行排序&#xff0c;最后将这些桶按顺序合并以得到排好序的数据集。 桶排序原理 确定桶的数量&am…
阅读更多...
Unity中用序列化和反序列化来保存游戏进度

Unity中用序列化和反序列化来保存游戏进度

[System.Serializable]标记类 序列化 [System.Serializable]是一个C#语言中的属性&#xff0c;用于标记类&#xff0c;表示该类的实例可以被序列化和反序列化。序列化是指将对象转换为字节流的过程&#xff0c;以便可以将其保存到文件、数据库或通过网络传输。反序列化则是将字…
阅读更多...
C++桶排序算法的应用：存在重复元素 III

C++桶排序算法的应用：存在重复元素 III

题目 给你一个整数数组 nums 和两个整数 indexDiff 和 valueDiff 。 找出满足下述条件的下标对 (i, j)&#xff1a; i ! j, abs(i - j) < indexDiff abs(nums[i] - nums[j]) < valueDiff 如果存在&#xff0c;返回 true &#xff1b;否则&#xff0c;返回 false 。 示例…
阅读更多...
日志技术快速入门

日志技术快速入门

1、创建Maven项目 这里不再说如何创建Maven项目 2、导入相关依赖 <dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.12</version></dependency>3、创建配置文件 在re…
阅读更多...
嵌入式系统学习路径：

嵌入式系统学习路径：

嵌入式系统学习路径&#xff1a; 00001. 确保扎实的C语言基础&#xff0c;包括高级编程知识和数据结构算法。 00002. 00003. 学习Linux应用层开发&#xff0c;包括并发程序设计、网络编程和数据库开发。 00004. 00005. 探索无线通信领域&#xff0c;如Zigbee、低功…
阅读更多...
【java学习—八】对象类型转换Casting（1）

【java学习—八】对象类型转换Casting（1）

文章目录 1. 数据类型转换1.1 基本数据类型的 Casting1.2. 对 Java 对象的强制类型转换(造型)2. 对象类型转换举例 1. 数据类型转换 数据类型转换分为基本数据类型转换和对象类型转换。 1.1 基本数据类型的 Casting (1) 自动类型转换&#xff1a;小的数据类型可以自动转换成…
阅读更多...
使用eBPF加速阿里云服务网格ASM

使用eBPF加速阿里云服务网格ASM

背景 随着云原生应用架构的快速发展&#xff0c;微服务架构已经成为了构建现代应用的主要方式之一。而在微服务架构中&#xff0c;服务间的通信变得至关重要。为了实现弹性和可伸缩性&#xff0c;许多组织开始采用服务网格技术来管理服务之间的通信。 Istio作为目前最受欢迎的…
阅读更多...
mac虚拟机安装homebrew时的问题

mac虚拟机安装homebrew时的问题

安装了mac虚拟机&#xff0c;结果在需要通过“brew install svn”安装svn时&#xff0c;才注意到没有下载安装homebrew。 于是便想着先安装homebrew&#xff0c;网上查的教程大多是通过类似以下命令 “ruby <(curl -fsSkL raw.github.com/mxcl/homebrew/go)” 但是都会出现…
阅读更多...
Mac OS m1 下安装Gradle5.1

Mac OS m1 下安装Gradle5.1

1. 下载、解压 1.1 下载地址 https://gradle.org 往下翻 选择 5.1 或者选择 任何 你想要的版本 ,点击 binary-only 即可下载 . 1.2 解压到指定目录 2. 配置环境变量 2.1 编辑环境文件 vi ~/.bash_profile #GRADLE相关配置 GRADLE_HOME/Users/zxj/Documents/devSoft/grad…
阅读更多...
c语言小白如何入门？

c语言小白如何入门？

c语言小白如何入门&#xff1f; 作为过来人&#xff0c;我觉得刚开始&#xff0c;先按照课本把每个知识点都弄懂&#xff0c;有不懂的地方&#xff0c;先尝试自己理解或借助互联网先搜一下&#xff0c;还是理解不了&#xff0c;就可以去找学得比较好的同学&#xff0c; 最近很…
阅读更多...
HarmonyOS 远端状态订阅开发实例

HarmonyOS 远端状态订阅开发实例

IPC/RPC 提供对远端 Stub 对象状态的订阅机制&#xff0c; 在远端 Stub 对象消亡时&#xff0c;可触发消亡通知告诉本地 Proxy 对象。这种状态通知订阅需要调用特定接口完成&#xff0c;当不再需要订阅时也需要调用特定接口取消。使用这种订阅机制的用户&#xff0c;需要实现消…
阅读更多...
【Leetcode刷题（数据结构）】：三路划分与三数随机取中的思想实现快速排序的再优化

【Leetcode刷题（数据结构）】：三路划分与三数随机取中的思想实现快速排序的再优化

快速排序是Hoare于1962年提出的一种二叉树结构的交换排序方法&#xff0c;其基本思想为&#xff1a;任取待排序元素序列中 的某元素作为基准值&#xff0c;按照该排序码将待排序集合分割成两子序列&#xff0c;左子序列中所有元素均小于基准值&#xff0c;右 子序列中所有元素均…
阅读更多...
Asp.net core Web Api 配置swagger中文

Asp.net core Web Api 配置swagger中文

启动项目&#xff0c;如图&#xff1a; 原来是英文的&#xff0c;我们要中文的&#xff0c;WeatherForecastController.cs是一个示例&#xff0c;删除即可&#xff0c;WeatherForecast.cs同时删除&#xff0c;当然不删除也行&#xff0c;这里是删除&#xff0c;创建自己的控制器…
阅读更多...
在不安全的集群上启用 Elasticsearch Xpack 安全性

在不安全的集群上启用 Elasticsearch Xpack 安全性

本博文详细描述如何把一个没有启动安全的 Elasticsearch 集群升级为一个带有 HTTPS 访问的启用 Elasticsearch xpack 安全的集群。 为了增强 Elasticsearch 集群的安全性&#xff0c;你需要执行完全集群重启&#xff0c;并在客户端进行一些更改。 启用身份验证后&#xff0c;所…
阅读更多...
linux环境下使用lighthouse与selenium

linux环境下使用lighthouse与selenium

一、安装谷歌浏览器、谷歌浏览器驱动、lighthouse shell脚本 apt update && apt -y upgrade apt install -y curl curl -fsSL https://deb.nodesource.com/setup_18.x | bash apt install -y nodejs apt install -y npm npm install -g lighthouse apt-get install -y …
阅读更多...
vue 01

vue 01

安装vscode 按照如下方式汉化 给vscode 安装插件 Vetur 安装node.js 下载地址https://nodejs.org/en/download/&#xff0c;选择windows msi 在cmd下检查如下&#xff1a; 检查nodejs版本:node --version 检查npm的安装版本&#xff1a;npm -v 执行命令&#xff1a; npm i…
阅读更多...
【机器学习】PyTorch-MNIST-手写字识别

【机器学习】PyTorch-MNIST-手写字识别

文章目录 前言完成效果一、下载数据集手动下载代码下载MNIST数据集&#xff1a; 二、 展示图片三、DataLoader数据加载器四、搭建神经网络五、 训练和测试第一次运行&#xff1a; 六、优化模型第二次优化后运行&#xff1a; 七、完整代码八、手写板实现输入识别功能 前言 注意…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023