flink学习：侧输出

什么是侧输出

在flink处理数据流时，我们经常会遇到这样的情况：在处理一个数据源时，往往需要将该源中的不同类型的数据做分割处理，如果使用 filter算子对数据源进行筛选分割的话，势必会造成数据流的多次复制，造成不必要的性能浪费；flink中的侧输出就是将数据流进行分割，而不对流进行复制的一种分流机制。flink的侧输出的另一个作用就是对延时迟到的数据进行处理，这样就可以不必丢弃迟到的数据。

侧输出案例

我们结合实际案例说明一下flink侧输出的用法，假设我们的需求是实时的从kafka接收生产数据，我们需要对迟到超过一定时长的数据另行处理：

定义数据类

case class kafkaData(
                      id: String, //id
                      eventTime: Long // 时间时间
                    ) {

  val delayTime = System.currentTimeMillis() / 1000 - eventTime

}

object kafkaData {
  def apply(str: String): kafkaData = {
    val strings = str.split(",")
    kafkaData(strings.head, strings(1).toLong)
  }
}

flink流处理程序

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 从Kafka获取数据流
    val kafkaConsumer = KafkaConsumerSource.getComsumer(
      load.getString("kafkaBootstrap"),
      "test",
      load.getString("kafkaGroupId_weather"),
      new SimpleStringSchema
    )

    kafkaConsumer.setStartFromLatest()

第一步：定义OutputTag

    // 定义 OutputTag 侧输出的数据格式可以不应和主流的数据格式一样
    val delayOutputTag = OutputTag[String]("delay-side-output")

第二步：使用特定的函数将数据发送到侧输出
使用侧输出时需要使用特定的函数发送数据，具体可以使用一下函数：

• ProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

  val dataSource = env.addSource(kafkaConsumer)
      .map(kafkaData.apply(_))
       //使用特定的函数将数据发送到侧输出
      .process(new ProcessFunction[kafkaData, kafkaData] {
        override def processElement(value: kafkaData,
                                    ctx: ProcessFunction[kafkaData, kafkaData]#Context,
                                    out: Collector[kafkaData]) = {
          if (value.delayTime < 100) {
            // 数据常规输出
            out.collect(value)
          } else {
            // 数据侧输出
            ctx.output(delayOutputTag, s"数据 ${value.toString} 迟到了 ：" + value.delayTime + "秒")
          }
        }
      })

    // 常规数据处理
    dataSource.print()

    // 对侧输出的数据处理
    dataSource.getSideOutput(delayOutputTag).print()
    
    env.execute("Side Outputs Test")

  }

我们往kafka中写入两条数据

1,1560522012
1,1560522698

程序输出结果：对不同分割源的数据做了不同的处理

4> 数据 kafkaData(1,1560522012) 迟到了 ：697秒
3> kafkaData(1,1560522698)

---

flink中也提供了专对迟到数据侧输出的方法：sideOutputLateData
使用方式：

 val delayOutputTag = OutputTag[T]("delay-side-output")

 val maindata  =input
       .keyBy(<key selector>)
      .window(<window assigner>)
      .sideOutputLateData(delayOutputTag)

val delayDataStream = ,maindata.getSideOutput(delayOutputTag)

参考资料

https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/side_output.html