Flink数据流类型之间的转换(WindowedStream、DataStream、KeyedStream、AllWindowStream之间的转换)

article/2025/9/16 19:41:16

       Flink提供了一些流API,其中包括WindowedStream、DataStream、KeyedStream和AllWindowStream。

🍊WindowedStream是一种特殊的流,其中数据已按时间或数据元素的键进行分组,并且每个分组的数据都在窗口中按时间划分。这意味着,如果你有一个WindowedStream,你可以对每个窗口执行转换,例如聚合或统计。

🍊DataStream是Flink中最基本的流类型,表示一个无界的、有序的数据流。它可以是任何类型的数据,例如数值、字符串或复杂的对象。

🍊KeyedStream是一种特殊的DataStream,其中数据已按照一个键(通常是一个数值或字符串)进行分组。这意味着你可以对每个键执行转换,例如聚合或计数。

🍊AllWindowStream是一种特殊的WindowedStream,其中数据流被分成固定大小的所有窗口。这意味着你可以对整个数据流执行转换,而无需将数据分组。

如下图所示,WindowedStream、DataStream、KeyedStream、AllWindowStream之间的转换

在这里插入图片描述

~下面使用代码做一些简单的转换示例,希望能对你有所帮助

       如,你可以使用keyBy()函数将DataStream转换为KeyedStream。( DataStream -> KeyedStream)

DataStream<String> dataStream = ...;
//DataStream -> KeyedStream
KeyedStream<String, String> keyedStream = dataStream.keyBy(new KeySelector<String, String>() {@Overridepublic String getKey(String value) throws Exception {return value;}
});

       要将KeyedStream转换为WindowedStream,你可以使用window()函数。例如,以下代码将每个数据元素的键的流分成5秒的滑动窗口( KeyedStream-> WindowedStream):

KeyedStream<String, String> keyedStream = ...;
//KeyedStream-> WindowedStream
WindowedStream<String, String, TimeWindow> windowedStream = keyedStream.window(SlidingTimeWindows.of(Time.seconds(5)));

       还可以使用windowAll()函数将DataStream转换为AllWindowStream(DataStream-> AllWindowStream)。例如,以下代码将数据流分成10秒的滑动窗口:

DataStream<String> dataStream = ...;
//DataStream-> AllWindowStream
AllWindowStream<String, TimeWindow> allWindowStream = dataStream.windowAll(SlidingTimeWindows.of(Time.seconds(10)));

       你可以使用以下代码将WindowedStream转换为DataStream(WindowedStream-> DataStream):

WindowedStream<T> windowedStream = ...;
DataStream<T> dataStream = windowedStream.windowAll(TumblingEventTimeWindows.of(Time.milliseconds(10)));

       你可以使用reduce函数将KeyedStream转换为DataStream(KeyedStream-> DataStream)。例如,假设你有一个整数类型的KeyedStream,并希望将其转换为所有键的和的DataStream,你可以使用以下代码:

KeyedStream<Integer, String> keyedStream = ...;
DataStream<Integer> sumStream = keyedStream.reduce(new ReduceFunction<Integer>() {public Integer reduce(Integer value1, Integer value2) {return value1 + value2;}
});

       你可以使用以下代码将DataStream转换为WindowedStream(DataStream-> WindowedStream)。这段代码将DataStream转换为带有滑动窗口的KeyedStream,然后使用window函数将其转换为WindowedStream,最后使用WindowFunction将WindowedStream中的数据进行转换。

DataStream<T> dataStream = ...;
WindowedStream<T, K, TimeWindow> windowedStream = dataStream.keyBy(new KeySelector<T, K>() {public K getKey(T value) {// Return the key for the value}
}).window(SlidingEventTimeWindows.of(Time.milliseconds(10), Time.milliseconds(5))).apply(new WindowFunction<T, T, K, TimeWindow>() {public void apply(K key, TimeWindow window, Iterable<T> values, Collector<T> out) {for (T value : values) {out.collect(value);}}});

http://chatgpt.dhexx.cn/article/WbgmZqJC.shtml

相关文章

DataStream API

目录 原算子 准备工作&#xff0c;环境搭建 读取数据 从文件中读取数据 从集合中读取数据 从元素中读取数据 从source文件中读取数据 从kafka中读取数据 自定义source类型输出 转换算子 map转换 Filter转换 FlatMap转换 原算子 准备工作&#xff0c;环境搭建 为…

Flink学习——DataStream API

一个flink程序&#xff0c;其实就是对DataStream的各种转换。具体可以分成以下几个部分&#xff1a; 获取执行环境&#xff08;Execution Environment&#xff09;读取数据源&#xff08;Source&#xff09;定义基于数据的转换操作&#xff08;Transformations&#xff09;定义…

大数据开发-Flink-数据流DataStream和DataSet

文章目录 一、DataStream的三种流处理Api1.1 DataSource1.2 Transformation1.3 Sink 二、DataSet的常用Api2.1 DataSource2.2 Transformation2.3 Sink Flink主要用来处理数据流&#xff0c;所以从抽象上来看就是对数据流的处理&#xff0c;正如前面大数据开发-Flink-体系结构 &…

Flink DataStream API 介绍

Flink DataStream API 介绍 StreamExecutionEnvironment #mermaid-svg-JKeWa22W2vWA4zBS {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-JKeWa22W2vWA4zBS .error-icon{fill:#552222;}#mermaid-svg-JKeWa22W2vWA4z…

DataStream API介绍与使用(一)

详细API参考官网 DataStream编程模型 在Flink整个系统架构中&#xff0c;对流计算的支持是其最重要的功能之一&#xff0c;Flink基于Google提出的DataFlow模型&#xff0c;实现了支持原生数据流处理的计算引擎。Flink中定义了DataStream API让用户灵活且高效地编写Flink流式应…

DataStream API(一)

Flink 有非常灵活的分层 API 设计&#xff0c;其中的核心层就是 DataStream/DataSet API。由于新版 本已经实现了流批一体&#xff0c; DataSet API 将被弃用&#xff0c;官方推荐统一使用 DataStream API 处理流数 据和批数据。由于内容较多&#xff0c;我们将会用几章的篇幅来…

DataStream(二)

目录 5.3.2 聚合算子&#xff08;Aggregation&#xff09; 5.3.3 用户自定义函数&#xff08;UDF&#xff09; 3. 扁平映射&#xff08;flatMap&#xff09; flatMap 操作又称为扁平映射&#xff0c;主要是将数据流中的整体&#xff08;一般是集合类型&#xff09;拆分成一个 …

Flink DataStream API

Flink DataStream API 编程指南 概览前言什么是DataStreamFlink程序剖析程序样例 Data SourcesDataStream Transformations算子数据流转换算子物理分区算子链和资源组 Data Sinks迭代执行参数 概览 前言 Flink中的DataStream程序是常规程序&#xff0c;可对数据流进行转换&am…

DataStream API(三)

目录 5.3.4 物理分区&#xff08;Physical Partitioning&#xff09; 5.4 输出算子&#xff08;Sink&#xff09; 5.4.1 连接到外部系统 5.4.2 输出到文件 5.4.3 输出到 Kafka 5.4.4 输出到 MySQL&#xff08;JDBC&#xff09; 5.4.5 自定义 Sink 输出 5.5 本章总结 5.3.…

流式数据采集和计算(十):Flink的DataStream学习笔记

Flink的DataStream学习笔记.. 1 Flink 基础.. 3 Flink特性.. 3 Flink和Spark对比.. 3 设计思路.. 3 状态管理.. 3 Flink 初探.. 4 设计架构.. 4 Flink on yarn. 5 流程分析.. 6 DataStream. 7 API程序结构.. 7 DataSource 8 Transformation. 9 Sink. 13 Time 14…

DataStream API(基础篇) 完整使用 (第五章)

DataStream API基础篇 一、执行环境&#xff08;Execution Environment&#xff09;1、创建执行环境1. getExecutionEnvironment2. createLocalEnvironment3. createRemoteEnvironment 二、执行模式(Execution Mode)1. BATCH模式的配置方法&#xff08;1&#xff09;通过命令行…

DataStream API 四 之 Flink DataStream编程

DataStream API 四 之 Flink DataStream编程 1.分布式流处理基本模型2.流应用开发步骤3.数据类型4. Connector5. Execution environment6. 参数传递7.配置并⾏度8.Watermark9.Checkpoint10.State11. Data Source11.111.2 自定义Source 12.Transformations13.Window13.1窗⼝处理…

Flink的DataStream介绍

1|0一&#xff1a;流式处理基本概念 流处理系统本身有很多自己的特点。一般来说&#xff0c;由于需要支持无限数据集的处理&#xff0c;流处理系统一般采用一种数据驱动的处理方式。它会提前设置一些算子&#xff0c;然后等到数据到达后对数据进行处理。 为了表达复杂的逻辑&am…

Flink DataStream API(基础版)

概述 DataStream&#xff08;数据流&#xff09;本身是 Flink 中一个用来表示数据集合的类&#xff08;Class&#xff09;&#xff0c;我们编写的 Flink 代码其实就是基于这种数据类型的处理&#xff0c;所以这套核心API 就以DataStream 命名。对于批处理和流处理&#xff0c;我…

node.js上开启服务,在同一局域网下的另一客户端访问

选择的服务是我之前做的案例&#xff1a;链接 1.在本机上开启服务&#xff1a; 2.本机上用浏览器访问验证无误&#xff1a; 3.运行cmd使用命令ipconfig查看本机ip地址 4.在另一台局域网下的机子&#xff0c;要求可以ping到。 浏览器访问ip地址&#xff1a;3000即可。&#…

前端 面试题

介绍项目 安全性 token 验证 处理令牌续期问题&#xff0c;在header中获取到新令牌时&#xff0c;替换老令牌&#xff0c;以达到用户无感刷新令牌 1、第一次登录的时候&#xff0c;前端调后端的登陆接口&#xff0c;发送用户名和密码 2、后端收到请求&#xff0c;验证用户名和…

基础知识---cmd命令行篇

1、echo&#xff08;输出&#xff09; > 覆盖 >>追加 2、dir&#xff08;展示当前目录的文件 .当前的目录 ..表示上一层目录&#xff09; 3、d&#xff1a;和cd&#xff1a;d为驱动器 cd展示当前目录的文件&#xff08; .为当前文件 ..为上一层目录的文件&#xf…

虚拟机NAT+静态IP+DNS

NAT模式下 虚拟机联网是通过物理机的VMware Nat服务&#xff08;电脑网络切换也无碍&#xff09;&#xff0c;禁用状态下ping不通物理机&#xff0c;也连不了网 物理机连接虚拟机的通过VMnet8虚拟网络适配器&#xff0c;禁用情况下ping不通虚拟机&#xff0c;Xshell工具也没法…

ACP考前错题总结(精华,已过ACP)

前言 证书和战绩镇楼&#xff0c;希望大家都可以拿到自己想要的Certificate。无论ACA-ACP-ACE亦或者GCP、AWS等等。 错题总结 镜像、本地磁盘部分 不建议基于本地服务器制作镜像上传到阿里云ECS并提供服务。 不支持写在座位数据盘使用的本地盘 镜像和快照&#xff1a;镜像可…

React+Native Unable to download JS Bundle解决办法

在配置ReactNative开发环境中&#xff0c;会遇到很多坑。 这个会困扰很多很多人。 在前序工作中&#xff0c;我们开启了8081端口&#xff0c;以供手机通过该端口下载相应的js。 而在命令行执行adb reverse tcp:8081 tcp:8081命令能解决一定的问题&#xff0c;但也有可能失效。…