【Flink快速入门-5.流处理之多流转换算子】
流处理之多流转换算子
实验介绍
前面实验中介绍的算子已经能够满足我们的大部分开发需求了，但是在实际工作中有时候还会遇到一些业务场景，例如需要摄入多个输入流并将其合并处理，或者需要将一条输入流分割为多条子流，在不同的子流中处理不同的业务逻辑。所以本节实验的内容我们将学习 DataSteam API 中的可以将多条输入流合并为一个输入流，或者将一个输入流分割为多个子流的算子，我们将其统称为“多流转换算子”。
知识点
- Union
- filter
算子演示
Union
union 顾名思义就是连接的意思，所以 union 算子的作用就是合并两条或者多条相同类型的 DataStream，生成一个新的类型相同的 DataStream。如图所示：
需要注意的是，事件合流的方式为 FIFO 方式。操作符并不会产生一个特定顺序的事件流。union 操作符也不会进行去重。每一个输入事件都被发送到了下一个操作符。
假设某公司分别在淘宝和天猫都开设了自己的直营店，公司高层需要实时监控到两个店铺的交易数据，并希望通过大屏展示的方式实时滚动。我们可以通过两条 Socket 输入流来模拟这样的场景。
首先在我们 FlinkLearning 工程的 com.vlab.operator 包下创建一个 UnionOperator 的 Scala object，输入如下代码：
package com.vlab.operator
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
object UnionOperator {
def main(args: Array[String]): Unit = {
// 创建执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 接收京东订单
val jdOrder:DataStream[String] = env.socketTextStream("192.168.137.81", 9999)
// 接收拼刀刀订单
val pindaoOrder:DataStream[String] = env.socketTextStream("192.168.137.81", 9998)
// 将两条输入流合并为一条输入流
val unionStream:DataStream[String] = jdOrder.union(pindaoOrder)
// 设置并行度
unionStream.print().setParallelism(1)
// 执行
env.execute("UnionOperator")
}
}
我们使用 netcat 监控两个端口来模拟发送淘宝和天猫的订单信息，然后使用 Flink 接收。打开终端窗口，执行 nc -l -p 9998 命令，紧接着打开另一个终端窗口，执行 nc -l -p 9999 命令。这样的话我们监控了 9998 和 9999 两个端口，接下来在 Flink 中进行接收。
运行刚刚的代码，然后在前面打开的两个终端中交替发送订单数据，观察 idea 控制台输出。
filter
使用 filter 来根据体温的阈值将流拆分为两个子流：一个是正常体温流，另一个是发烧体温流。然后我们可以对每个子流进行不同的业务逻辑处理。
疫情期间，全国各地的超市、医院、机场等公共场所入口都有温度监控设备，当该设备检测到某个人体温异常之后就会报警。假设鉴别正常体温和发烧体温的阈值为 36.0 摄氏度，也就是说，只要体温大于等于 36.0 摄氏度我们就认为其为发烧状态。我们使用 检测体温是否异常，我们可以使用 filter 来将流分为两条子流，一个代表 正常体温，另一个代表 发烧体温，然后可以对这些流进行不同的业务逻辑处理。
在我们 FlinkLearning 工程的 com.vlab.operator 包下创建一个名为 SelectOperator 的 Scala object，代码如下：
package com.shiyanlou.operator
import org.apache.flink.streaming.api.scala._
object SelectOperator {
def main(args: Array[String]): Unit = {
// 设置流环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 读取socket文本数据流
val inputDS: DataStream[String] = env.socketTextStream("192.168.137.81", 9999)
val peopleStream = inputDS
.map(line => {
val arr = line.split(" ")
People(arr(0), arr(1).toFloat)
})
// 使用 keyBy 按照温度类型（high 或 normal）进行分组
val highTempStream = peopleStream.filter(_.temperature > 36.5)
val normalTempStream = peopleStream.filter(_.temperature <= 36.5)
// 打印输出
highTempStream.print("发烧")
normalTempStream.print("体温正常")
env.execute("SelectOperator")
}
case class People(name: String, temperature: Float)
}
上面的代码中，我们创建了一个 Socket 输入流监控localhost下的 9999 端口，然后将输入的文本使用空格分隔之后转换为People类。紧接着使用 Split 算子将体温大于 36.0 的人群定义为fever，将体温小于等于 36.0 的人群定义为normal，最后使用select算子选择了fever（发烧）状态的人群并输出到控制台。
打开终端，执行nc -l -p 9999，在 idea 运行以上代码，并在终端中依次发送下面的信息：
张小明 35.6
李鹏程 36.3
赵露 36.7
李阳 35.5
刘明 37.0
在 idea 的控制台会看到将体温高于 36.5 的做了打印（赵露、刘明）。
实验总结
本节实验中我们介绍了 Flink 中的多流转换算子，其中 Union 是将两个或者多个类型相同的输入流转换成一个输入流，而filter是将一个输入流根据给定的条件切分成多个子输入流。这部分内容在工作中会经常用到，大家一定要理解。