Flink DataStream API之Operators

2023-12-06 03:08
文章标签 api flink operators datastream

本文主要是介绍Flink DataStream API之Operators，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Flink DataStream API之Operators

官网位置：https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/operators/

Operators transform one or more DataStreams into a new DataStream. Programs can combine multiple transformations into sophisticated dataflow topologies.

算子可以将一个或多个DataStream转化为新的DataStream。程序可以将多种转换组合成复杂的数据流拓扑。下面将介绍一下Flink DataStream API中一些常用的算子，注意：本文是以Scala代码为例，有些语法在Java里不通用。

1 DataStream Transformations

官网以表格的形式列了一些转换类的算子，有些可能不好理解，这里以实际的代码例子进行演示。

DataStream常用转换如下图所示：

1.1 map

类型转换：

DataStream->DataStream

描述：

熟悉scala的同学，对于这个map应该很好理解，map操作是依次取DataStream中一个元素并生成另一个元素。

调用方式:

  1. 传入一个继承自MapFunction[T,R]的实例
def map[R: TypeInformation](mapper: MapFunction[T, R]): DataStream[R]
  1. 函数式调用，传入一个参数类型为T，返回值为R的函数
  def map[R: TypeInformation](fun: T => R): DataStream[R]

代码演示：

先通过1到10的集合生成一个DataStream，并通过map操作，将其每个元素都乘以2，生成新的DataStream。

    // create DataStream from collectionval data = env.fromCollection(1 to 10)// 将每个元素*2data.map(_ * 2).print("map")

输出：

map:7> 10
map:4> 4
map:3> 2
map:2> 16
map:8> 12
map:6> 8
map:5> 6
map:1> 14
map:3> 18
map:4> 20

1.2 flatMap

类型转换：

DataStream->DataStream

描述：

flatMap与map类似，只是它是依次取DataStream中的一个元素生成0个或多个元素。

调用方式：

它在scala里提供了三种调用方式：

  1. 传入一个继承自FlatMapFunction[T, R]的类实例
def flatMap[R: TypeInformation](flatMapper: FlatMapFunction[T, R]): DataStream[R]
  1. 传入参数为T和Collector[R]无返回值的函数
 def flatMap[R: TypeInformation](fun: (T, Collector[R]) => Unit): DataStream[R]
  1. 传入参数类型为T，返回值类型为TraversableOnce(集合特质，常见集合都继承该特质)的函数
def flatMap[R: TypeInformation](fun: T => TraversableOnce[R]): DataStream[R]

代码演示：

先通过Array(1,2,3)和Array(4,5,6)两个元素生成一个DataStream，并通过flatMap操作，将两个数组展开

    // create DataStream from elementsval data = env.fromElements(Array(1,2,3),Array(4,5,6))// 将数据展开data.flatMap(_.toList).print("flatMap")

结果：

flatMap:1> 4
flatMap:8> 1
flatMap:1> 5
flatMap:8> 2
flatMap:1> 6
flatMap:8> 3

1.3 filter

类型转换：

DataStream->DataStream

描述：

按照条件过滤，得到满足条件的元素生成新的DataStream

调用方式：

调用方式有两种：

  1. 传入一个继承自FilterFunction[T]类实例
def filter(filter: FilterFunction[T]): DataStream[T]
  1. 传入一个入参类型为T，返回值类型为Boolean的函数
def filter(fun: T => Boolean): DataStream[T]

代码演示：

先通过1 到10的集合生成一个DataStream，并通过filter操作，过滤出奇数

    // create DataStream from collectionval data = env.fromCollection(1 to 10)// 输出奇数data.filter(_ % 2 == 1).print("filter")

输出：

filter:3> 1
filter:5> 3
filter:7> 5
filter:1> 7
filter:3> 9

1.4 keyBy

类型转换：

DataStream->KeyedStream

描述：

根据指定的Key选择器，进行分区，生成KeyedStream

调用方式：

在scala里key选择器有四种指定方式：

  1. 对于tuple和array类型，我们可以指定一个或多个key的位置
def keyBy(fields: Int*): KeyedStream[T, JavaTuple]
  1. 指定一个或多个key的字段表达式
def keyBy(firstField: String, otherFields: String*): KeyedStream[T, JavaTuple]
  1. 传入key选择函数
def keyBy[K: TypeInformation](fun: T => K): KeyedStream[T, K]
  1. 实现key选择器KeySelector，并传入该实例
def keyBy[K: TypeInformation](fun: KeySelector[T, K]): KeyedStream[T, K]

代码演示：

先通过集合生成一个DataStream，然后根据第一个元素进行KeyBy

    // create DataStream from collectionval data = env.fromCollection(List(("device-1", 1), ("device-2", 2), ("device-1", 2)))data.keyBy(_._1)

输出：

keyed:3> (device-2,2)
keyed:1> (device-1,1)
keyed:1> (device-1,2)

1.5 reduce

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，将当前值以及上次计算结果应用到指定的reduce函数中得到新的值，依次计算

调用方式：

在scala里reduce调用有两种方式：

  1. 实现ReduceFunction[T]接口中的reduce方法，并传入该实例
 def reduce(reducer: ReduceFunction[T]): DataStream[T]
  1. 传入参数类型为T,T返回值为T的函数
def reduce(fun: (T, T) => T): DataStream[T]

代码演示：

将上面KeyBy之后生成的KeyedStream，应用reduce算子，进行累加

    keyedStream.reduce((value1, value2) => {(value1._1, value1._2 + value2._2)}).print("reduce")

输出：

reduce:1> (device-1,1)
reduce:3> (device-2,2)
reduce:1> (device-1,3)

1.6 fold(deprecated)

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，与reduce类似，只是多了一个初始值。

调用方式：

在scala里fold调用有两种方式：

  1. 实现FoldFunction[T,R]接口中的fold方法，并传入该实例,另外还需要传入初始化值。
  def fold[R: TypeInformation](initialValue: R, folder: FoldFunction[T,R]): DataStream[R]
  1. 柯里化传参，第一个参数为初始值，第二个参数为函数表达式
def fold[R: TypeInformation](initialValue: R)(fun: (R,T) => R): DataStream[R]

代码演示：

将上面KeyBy之后生成的KeyedStream，应用fold算子，进行字符串拼接

    keyedStream.fold("start")((accumulator, value) => {s"$accumulator-${value._1}-${value._2}"}).print("fold")

输出：

fold:1> start-device-1-1
fold:3> start-device-2-2
fold:1> start-device-1-1-device-1-2

1.7 aggregations

类型转换：

KeyedStream->DataStream

描述：

在KeyedStream上，应用聚合方法，Flink提供内置的聚合方法有：sum、min、minBy、maxBy。这里解释一下min和minBy，max和maxBy之间的却别。以max和maxBy为例，max(position|field)只是返回指定的position或field的最大值，其它字段的值为第一次出现的记录值，而maxBy(position|field)则是返回position或field最大值所在的记录。怎么理解呢？假如streamData:DataStream[DeviceEvent]有两条记录 DeviceEvent(1,1,1)、 DeviceEvent(2,2,2)，即

val streamData:DataStream[DeviceEvent]=env.fromElements(DeviceEvent(1,1,1),DeviceEvent(2,2,2))

我们通过按照第一个元素keyBy，然后按照第二个元素max

streamData.keyBy(0).max(1)

我们会得到结果：

1,1,1
1,2,1

如果我们同样按照第一个元素keyBy，但是按照第二个元素maxBy

streamData.keyBy(0).maxBy(1)

将会得到如下结果：

1,1,1
2,2,2

min和minBy同理。

调用方式：

flink内置聚合方法在scala里的调用方式有两种：

  1. 传入position参数
def sum(position: Int): DataStream[T]
  1. 传入field参数
def sum(field: String): DataStream[T]

代码演示：

演示这几种聚合方法

    // create DataStream from collectionval data: DataStream[(String, Int, String)] = env.fromCollection(List(("device-1", 3, "1-3"), ("device-1", 1, "1-1"), ("device-2", 2, "2-2"), ("device-1", 2, "1-2")))val keyedStream = data.keyBy(0)keyedStream.sum(1).print("sum")keyedStream.min(1).print("min")keyedStream.max(1).print("max")keyedStream.minBy(1).print("minBy")keyedStream.maxBy(1).print("maxBy")

输出：

sum:3> (device-2,2,2-2)
max:3> (device-2,2,2-2)
min:1> (device-1,3,1-3)
maxBy:3> (device-2,2,2-2)
maxBy:1> (device-1,3,1-3)
minBy:1> (device-1,3,1-3)
sum:1> (device-1,3,1-3)
max:1> (device-1,3,1-3)
minBy:3> (device-2,2,2-2)
min:3> (device-2,2,2-2)
max:1> (device-1,3,1-3)
minBy:1> (device-1,1,1-1)
maxBy:1> (device-1,3,1-3)
max:1> (device-1,3,1-3)
min:1> (device-1,1,1-3)
minBy:1> (device-1,1,1-1)
sum:1> (device-1,4,1-3)
min:1> (device-1,1,1-3)
maxBy:1> (device-1,3,1-3)
sum:1> (device-1,6,1-3)

1.8 window

类型转换：

KeyedStream->WindowedStream

描述：

在KeyedStream上的窗口装换，Flink提供了四种窗口类型：Tumbling Windows(滚动窗口)、Sliding Windows(滑动窗口)、Session Windows(会话窗口)、Global Windows(全局窗口)，滚动窗口和滑动窗口分为基于时间的窗口和基于个数的窗口，而基于时间的窗口又分为事件时间窗口和处理时间窗口。关于窗口的更多内容会在下一篇文章中记录，这里只做简单演示。

调用方式：

window在scala里的调用方式有五种：

  1. timeWindow(size: Time)，传入类型为Time的窗口大小，Flink会根据指定的time characteristic来判断是处理时间滚动窗口还是事件时间滚动窗口。
def timeWindow(size: Time): WindowedStream[T, K, TimeWindow]
  1. timeWindow(size: Time, slide: Time)，传入类型为Time的窗口大小，以及类型为Time的窗口滑动大小，Flink同样会根据指定的time characteristic来判断是处理时间滑动窗口还是事件时间滑动窗口。
def timeWindow(size: Time, slide: Time): WindowedStream[T, K, TimeWindow]
  1. countWindow(size: Long)，传入Long类型的窗口大小，Flink会应用为基于个数的滚动窗口。
def countWindow(size: Long): WindowedStream[T, K, GlobalWindow]
  1. countWindow(size: Long, slide: Long)，传入Long类型的窗口大小，以及Long类型的窗口滑动带下，生成基于个数的滑动窗口。
def countWindow(size: Long, slide: Long): WindowedStream[T, K, GlobalWindow]
  1. window，需要传入WindowAssigner实例
def window[W <: Window](assigner: WindowAssigner[_ >: T, W]): WindowedStream[T, K, W]

代码演示：

    // create data from collectionval data = env.fromCollection(1 to 10)data.countWindowAll(5)

1.9 windowAll

类型转换：

DataStream->AllWindowedStream

描述：

在DataStream上的窗口装换，与window类似，只是window应用在KeyedStream上，需要先进行KeyBy操作，windowAll可以直接应用到DataStream上。

调用方式：

window在scala里的调用方式有五种：

  1. timeWindowAll(size: Time)传入类型为Time的窗口大小，Flink会根据指定的time characteristic来判断是处理时间滚动窗口还是事件时间滚动窗口。
def timeWindowAll(size: Time): AllWindowedStream[T, TimeWindow]
  1. timeWindow(size: Time, slide: Time)，传入类型为Time的窗口大小，以及类型为Time的窗口滑动大小，Flink同样会根据指定的time characteristic来判断是处理时间滑动窗口还是事件时间滑动窗口。
def timeWindowAll(size: Time, slide: Time): AllWindowedStream[T, TimeWindow]
  1. countWindow(size: Long)，传入Long类型的窗口大小，Flink会应用为基于个数的滚动窗口。
def countWindowAll(size: Long, slide: Long): AllWindowedStream[T, GlobalWindow]
  1. countWindow(size: Long, slide: Long)，传入Long类型的窗口大小，以及Long类型的窗口滑动带下，生成基于个数的滑动窗口。
def countWindowAll(size: Long): AllWindowedStream[T, GlobalWindow]
  1. windowAll，需要传入WindowAssigner实例
windowAll[W <: Window](assigner: WindowAssigner[_ >: T, W]): AllWindowedStream[T, W]

代码演示：

    // create data from collectionval data = env.fromCollection((1 to 10).map(x => (x % 2, x)))// get KeyedDataStreamval keyedStream = data.keyBy(0)val windowStream = keyedStream.countWindow(5)

输出：

windowStream.sum(1).print("countWindow")

countWindow:6> (1,25)
countWindow:6> (0,30)

1.10 WindowedStream apply

类型转换：

WindowedStream->DataStream

描述：

在WindowedStream上应用WindowFunction，当窗口被触发时调用指定的WindowFunction。

调用方式：

apply在scala里的调用方式有两种(其它预聚合的调用方式已废弃)：

  1. 继承WindowFunction实现apply方法，并传入该实例
def apply[R: TypeInformation](function: WindowFunction[T, R, K, W]): DataStream[R]
  1. 传入匿名函数
def apply[R: TypeInformation](function: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R]

代码演示：

在上面生成的windowStream调用apply

    val dataStream = windowedStream.apply((key, window, input, out: Collector[(Int, String, Int)]) => {var sum = 0input.foreach(element => {sum += element._2})out.collect(key.getField(0), window.toString, sum)})dataStream.print("simple apply")

输出：

simple apply:6> (1,GlobalWindow,25)
simple apply:6> (0,GlobalWindow,30)

1.11 WindowedStream reduce

类型转换：

WindowedStream->DataStream

描述：

在WindowedStream上应用ReduceFunction，与DataStream上的ReduceFunction相同，当窗口被触发时调用指定的ReduceFunction。

调用方式：

reduce在scala里的调用方式有六种，其中包括几种预聚合的方式调用：

  1. 继承ReduceFunction实现reduce方法，并传入该实例
def reduce(function: ReduceFunction[T]): DataStream[T]
  1. 传入匿名函数
def reduce(function: (T, T) => T): DataStream[T]
  1. 传入ReduceFunction[T]作为预聚合器,当数据到达时调用，传入WindowFunction作为窗口函数，当触发窗口计算的时候调用。
  def reduce[R: TypeInformation](preAggregator: ReduceFunction[T],function: WindowFunction[T, R, K, W]): DataStream[R]
  1. 传入匿名函数作为预聚合器,当数据到达时调用，传入WindowFunction作为窗口函数，当触发窗口计算的时候调用。
  def reduce[R: TypeInformation](preAggregator: (T, T) => T,windowFunction: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R]
  1. 传入匿名函数作为预聚合器,当数据到达时调用，传入ProcessWindowFunction作为窗口函数，当触发窗口计算的时候调用。
  def reduce[R: TypeInformation](preAggregator: (T, T) => T,function: ProcessWindowFunction[T, R, K, W]): DataStream[R]
  1. 传入ReduceFunction[T]作为预聚合器,当数据到达时调用，传入ProcessWindowFunction作为窗口函数，当触发窗口计算的时候调用。
  def reduce[R: TypeInformation](preAggregator: ReduceFunction[T],function: ProcessWindowFunction[T, R, K, W]): DataStream[R]

代码演示：

在上面生成的windowStream调用reduce

    windowedStream.reduce(new ReduceSum).print("function reduce")windowedStream.reduce((value1, value2) => (value2._1, value1._2 + value2._2)).print("simple reduce")

输出：

simple reduce:6> (1,25)
function reduce:6> (1,25)
simple reduce:6> (0,30)
function reduce:6> (0,30)

1.11 WindowedStream fold(deprecated)

deprecated

1.12 WindowedStream aggregations

类型转换：

WindowedStream->DataStream

描述：

WindowedStream上的聚合函数，与DataStream上的相同，也有sum、min、minBy、max、maxBy，参考DataStream上的aggregations。

调用方式：

参考DataStream上的aggregations。

代码演示：

    windowedStream.sum(0).print("sum")windowedStream.min(0).print("min")windowedStream.max(0).print("max")windowedStream.minBy(0).print("minBy")windowedStream.maxBy(0).print("maxBy")

输出：

maxBy:6> (1,1)
minBy:6> (1,1)
max:6> (1,1)
sum:6> (5,1)
min:6> (1,1)
maxBy:6> (0,2)
minBy:6> (0,2)
max:6> (0,2)
sum:6> (0,2)
min:6> (0,2)

1.13 union

类型转换：

DataStream[T],DataStream[T],*->DataStream[T]

描述：

将两个或更多的流进行合并，要求数据类的类型是一直的。

调用方式：

该算子调用方式只有一种：

  1. 传入不定长的DataStream[T]
def union(dataStreams: DataStream[T]*): DataStream[T]

代码演示：

先通过1 到10的集合生成一个DataStream，并通过filter操作，过滤出奇数

    val stream1 = env.fromCollection(1 to 10)val stream2 = env.fromCollection(101 to 110)stream1.union(stream2).print("union")

输出：

union:4> 1
union:7> 101
union:4> 9
union:2> 104
union:3> 105
union:5> 107
union:5> 2
union:1> 103
union:6> 108
union:1> 6
union:6> 3
union:8> 102
union:5> 10
union:2> 7
union:4> 106
union:3> 8
union:7> 109
union:8> 110
union:7> 4
union:8> 5

1.13 join

类型转换：

DataStream[T],DataStream[T2]->JoinedStreams[T, T2]

描述：

将两个按照指定的条件流进行join，两个流的数据类型可以不同，需要与窗口函数结合使用。

调用方式：

该算子调用方式只有一种：

  1. 传入需要Join的DataStream[T2]
def join[T2](otherStream: DataStream[T2]): JoinedStreams[T, T2]

代码演示：

定义两个流，一个是从集合中创建，另一个从socket中创建。

    val stream1 = env.fromCollection(List(("device-1", 1), ("device-2", 2)))val stream2 = env.socketTextStream("localhost", 9090).map(x => {val token = x.split(" ")(token(0), token(1))})val joinedStream: JoinedStreams[(String, Int), (String, String)]#Where[String]#EqualTo = stream1.join(stream2).where(_._1).equalTo(_._1)joinedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10))).apply((first, second) => (first._1, second._2, first._2)).print("join")

使用nc -lk 9090监听9090端口，启动程序后，发送数据：

输出：

union:4> 1
union:7> 101
union:4> 9
union:2> 104
union:3> 105
union:5> 107
union:5> 2
union:1> 103
union:6> 108
union:1> 6
union:6> 3
union:8> 102
union:5> 10
union:2> 7
union:4> 106
union:3> 8
union:7> 109
union:8> 110
union:7> 4
union:8> 5

这篇关于Flink DataStream API之Operators的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/460185。

相关文章

基于Flask框架添加多个AI模型的API并进行交互

基于Flask框架添加多个AI模型的API并进行交互

《基于Flask框架添加多个AI模型的API并进行交互》：本文主要介绍如何基于Flask框架开发AI模型API管理系统,允许用户添加、删除不同AI模型的API密钥,感兴趣的可以了解下... 目录1. 概述2. 后端代码说明2.1 依赖库导入2.2 应用初始化2.3 API 存储字典2.4 路由函数2.5 应
阅读更多...
C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)

C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)

《C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)》本文主要介绍了C#集成DeepSeek模型实现AI私有化的方法,包括搭建基础环境,如安装Ollama和下载DeepS... 目录前言搭建基础环境1、安装 Ollama2、下载 DeepSeek R1 模型客户端 ChatBo
阅读更多...
Java调用DeepSeek API的最佳实践及详细代码示例

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》：本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.
阅读更多...
Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

《DeepseekR1模型本地化部署+API接口调用详细教程(释放AI生产力)》本文介绍了本地部署DeepSeekR1模型和通过API调用将其集成到VSCode中的过程,作者详细步骤展示了如何下载和... 目录前言一、deepseek R1模型与chatGPT o1系列模型对比二、本地部署步骤1.安装oll
阅读更多...
浅析如何使用Swagger生成带权限控制的API文档

浅析如何使用Swagger生成带权限控制的API文档

《浅析如何使用Swagger生成带权限控制的API文档》当涉及到权限控制时,如何生成既安全又详细的API文档就成了一个关键问题,所以这篇文章小编就来和大家好好聊聊如何用Swagger来生成带有... 目录准备工作配置 Swagger权限控制给 API 加上权限注解查看文档注意事项在咱们的开发工作里，API
阅读更多...
一分钟带你上手Python调用DeepSeek的API

一分钟带你上手Python调用DeepSeek的API

《一分钟带你上手Python调用DeepSeek的API》最近DeepSeek非常火,作为一枚对前言技术非常关注的程序员来说,自然都想对接DeepSeek的API来体验一把,下面小编就来为大家介绍一下... 目录前言免费体验API-Key申请首次调用API基本概念最小单元推理模型智能体自定义界面总结前言最
阅读更多...
JAVA调用Deepseek的api完成基本对话简单代码示例

JAVA调用Deepseek的api完成基本对话简单代码示例

《JAVA调用Deepseek的api完成基本对话简单代码示例》：本文主要介绍JAVA调用Deepseek的api完成基本对话的相关资料,文中详细讲解了如何获取DeepSeekAPI密钥、添加H... 获取API密钥首先，从DeepSeek平台获取API密钥，用于身份验证。添加HTTP客户端依赖使用Jav
阅读更多...
C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

《C#使用DeepSeekAPI实现自然语言处理,文本分类和情感分析》在C#中使用DeepSeekAPI可以实现多种功能,例如自然语言处理、文本分类、情感分析等,本文主要为大家介绍了具体实现步骤,... 目录准备工作文本生成文本分类问答系统代码生成翻译功能文本摘要文本校对图像描述生成总结在C#中使用Deep
阅读更多...
5分钟获取deepseek api并搭建简易问答应用

5分钟获取deepseek api并搭建简易问答应用

《5分钟获取deepseekapi并搭建简易问答应用》本文主要介绍了5分钟获取deepseekapi并搭建简易问答应用,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需... 目录1、获取api2、获取base_url和chat_model3、配置模型参数方法一：终端中临时将加
阅读更多...
使用DeepSeek API 结合VSCode提升开发效率

使用DeepSeek API 结合VSCode提升开发效率

《使用DeepSeekAPI结合VSCode提升开发效率》：本文主要介绍DeepSeekAPI与VisualStudioCode(VSCode)结合使用,以提升软件开发效率,具有一定的参考价值... 目录引言准备工作安装必要的 VSCode 扩展配置 DeepSeek API1. 创建 API 请求文件2.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2