Flink——Side Output侧输出流

比眉伴天荒 2022-10-31 04:29 193阅读 0赞

主要内容：

*  结合应用场景，介绍Flink侧输出流的使用流程和原理

在处理数据的时候，有时候想对不同情况的数据进行不同的处理，那么就需要把数据流进行分流。可以在主数据流上产生出任意数量额外的侧输出流。

## 1 场景 ##

某公司使用埋点组件收集到了埋点数据，并实时写入了Kafka。其中，埋点数据共分为三类：Web端埋点数据、移动端埋点数据和CS端埋点数据。现在需要从Kafka读取埋点数据，并分别对三端数据做不同的处理逻辑：

## 2 Side Output ##

当然使用 filter 对主数据流进行过滤，也能满足上述场景，但每次筛选过滤都要保留整个流，然后通过遍历整个流来获取相应的数据，显然很浪费性能。假如能够在一个流里面就进行多次输出就好了，恰好 Flink 的 Side Output 提供了这样的功能。Flink的Side Output侧输出流的作用在于将主数据分割成多个不同的侧输出流。侧输出结果流的数据类型不需要与主数据流的类型一致，不同侧输出流的类型也可以不同。

在上述场景中，可以使用Flink此功能：将Kafka的埋点数据进行分类，分为web端、mobile端和CS端三类，然后再对每类埋点数据进行相应的处理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1eHUyNA_size_16_color_FFFFFF_t_70_pic_center]

## 3 处理流程 ##

### 3.1 定义OutputTag ###

在使用侧输出的时候需要先定义一个OutputTag，来标识 Side Output，代表这个 Tag 是要收集哪种类型的数据。这里定义了三个 OutputTag：

*  `webTerminal`：web端埋点数据
 *  `mobileTerminal`：移动端埋点数据
 *  `csTerminal：CS`端埋点数据

lazy val webTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("Web端埋点数据")
    lazy val mobileTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("移动端埋点数据")
    lazy val csTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("CS端埋点数据")

### 3.2 使用特定的处理函数 ###

要使用侧输出，在处理数据的时候除了要定义相应类型的OutputTag外，还要使用特定的函数，主要是有四个：

*  `ProcessFunction`
 *  `CoProcessFunction`
 *  `ProcessWindowFunction`
 *  `ProcessAllWindowFunction`

该场景里选择使用ProcessFunction函数。该函数继承于RichFunction，使用时必须要重写processElement方法。自定义ProcessFunction函数：

### 3.3 对每个侧输出流做处理 ###

为每种类型的侧输出流添加处理逻辑，直接调用`getSideOutput`函数：（这里的处理逻辑进展示直接

## 4 代码 ##

### 4.1 代码 ###

package org.ourhome.streamapi
    
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.ProcessFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    /**
     * @Author Do
     * @Date 2020/5/3 20:35
     */
    object SideOutputTest2 {
      lazy val webTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("Web端埋点数据")
      lazy val mobileTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("移动端埋点数据")
      lazy val csTerminal: OutputTag[MdMsg] = new OutputTag[MdMsg]("CS端埋点数据")
    
      def main(args: Array[String]): Unit = {
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
        env.setParallelism(1)
    
        val socketData: DataStream[String] = env.socketTextStream("localhost", 9999)
        socketData.print("input data")
    
        val outputStream: DataStream[MdMsg] = socketData.map(line => {
          val str: Array[String] = line.split(",")
          MdMsg(str(0), str(1), str(2).toLong)
        })
          .process(new MdSplitProcessFunction)
    
        // Web端埋点数据流处理逻辑
        outputStream.getSideOutput(webTerminal).print("web")
        // Mobile端埋点数据流处理逻辑
        outputStream.getSideOutput(mobileTerminal).print("mobile")
        // CS端埋点数据流处理逻辑
        outputStream.getSideOutput(csTerminal).print("cs")
    
        env.execute()
    
      }
    
      case class MdMsg(mdType:String, url:String, Time:Long)
    
      class MdSplitProcessFunction extends ProcessFunction[MdMsg, MdMsg] {
    
        override def processElement(value: MdMsg, ctx: ProcessFunction[MdMsg, MdMsg]#Context, out: Collector[MdMsg]): Unit = {
          // web
          if (value.mdType == "web") {
            ctx.output(webTerminal, value)
            // mobile
          } else if (value.mdType == "mobile") {
            ctx.output(mobileTerminal, value)
            // cs
          } else if (value.mdType == "cs") {
            ctx.output(csTerminal, value)
            // others
          } else {
            out.collect(value)
          }
    
        }
      }
    
    }

### 4.2 测试 ###

输入1：

web,http://www.web1.com,1587787201000
    mobile,http://www.mobile1.com,1587787202000
    cs,http://www.cs1.com,1587787203000

输出1：

input data> web,http://www.web1.com,1587787201000
    web> MdMsg(web,http://www.web1.com,1587787201000)
    
    input data> mobile,http://www.mobile1.com,1587787202000
    mobile> MdMsg(mobile,http://www.mobile1.com,1587787202000)
    
    input data> cs,http://www.cs1.com,1587787203000
    cs> MdMsg(cs,http://www.cs1.com,1587787203000)

*  可见：Flink将接收到的数据，分别分到了web埋点数据流、移动端埋点数据流和CS端埋点数据流中，并分别打印。

输入2：

other,http://www.other.com,1587787201000

输出2：

input data> other,http://www.other.com,1587787201000

可见：Flink将接收到的数据发到了常规数据流中，并打印。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1eHUyNA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/af2e173ab0b94ffeb244e5bedb48f4b4.png